feat: add tdengine jdbc history writer

2026-06-29 13:28:36 +08:00
parent f58fe51cf5
commit 4de723d0b4
4 changed files with 394 additions and 0 deletions
--- 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
@@ -0,0 +1,147 @@
+package com.lingniu.ingest.tdenginehistory;
+
+import javax.sql.DataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter {
+
+    private final DataSource dataSource;
+    private final TdengineHistorySchema schema;
+    private final TdengineHistoryStatements statements;
+
+    public TdengineJdbcHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) {
+        if (dataSource == null) {
+            throw new IllegalArgumentException("dataSource must not be null");
+        }
+        if (schema == null) {
+            throw new IllegalArgumentException("schema must not be null");
+        }
+        this.dataSource = dataSource;
+        this.schema = schema;
+        this.statements = new TdengineHistoryStatements(schema);
+    }
+
+    public void initializeSchema() throws IOException {
+        inTransaction(connection -> {
+            try (Statement statement = connection.createStatement()) {
+                for (String sql : schema.bootstrapSql()) {
+                    statement.execute(sql);
+                }
+            }
+        });
+    }
+
+    @Override
+    public void appendRawFrames(List<TdengineRawFrameRow> rows) throws IOException {
+        append(rows, statements::rawFrame);
+    }
+
+    @Override
+    public void appendLocations(List<TdengineLocationRow> rows) throws IOException {
+        append(rows, statements::location);
+    }
+
+    private <T> void append(List<T> rows, Function<T, TdengineBatchStatement> mapper) throws IOException {
+        if (rows == null || rows.isEmpty()) {
+            return;
+        }
+        inTransaction(connection -> {
+            Set<String> createdChildTables = new LinkedHashSet<>();
+            Map<String, PreparedStatement> preparedStatements = new LinkedHashMap<>();
+            try {
+                for (T row : rows) {
+                    if (row == null) {
+                        continue;
+                    }
+                    TdengineBatchStatement batch = mapper.apply(row);
+                    if (createdChildTables.add(batch.createChildTableSql())) {
+                        try (Statement statement = connection.createStatement()) {
+                            statement.execute(batch.createChildTableSql());
+                        }
+                    }
+                    PreparedStatement prepared = preparedStatements.computeIfAbsent(batch.insertSql(), sql -> {
+                        try {
+                            return connection.prepareStatement(sql);
+                        } catch (SQLException e) {
+                            throw new JdbcRuntimeException(e);
+                        }
+                    });
+                    bind(prepared, batch.values());
+                    prepared.addBatch();
+                }
+                for (PreparedStatement prepared : preparedStatements.values()) {
+                    prepared.executeBatch();
+                }
+            } finally {
+                for (PreparedStatement prepared : preparedStatements.values()) {
+                    prepared.close();
+                }
+            }
+        });
+    }
+
+    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 void inTransaction(SqlWork work) throws IOException {
+        try (Connection connection = dataSource.getConnection()) {
+            boolean autoCommit = connection.getAutoCommit();
+            connection.setAutoCommit(false);
+            try {
+                work.execute(connection);
+                connection.commit();
+            } catch (JdbcRuntimeException e) {
+                rollback(connection);
+                throw e.getCause();
+            } catch (SQLException e) {
+                rollback(connection);
+                throw e;
+            } finally {
+                connection.setAutoCommit(autoCommit);
+            }
+        } catch (SQLException e) {
+            throw new IOException("tdengine history write failed", e);
+        }
+    }
+
+    private static void rollback(Connection connection) throws SQLException {
+        connection.rollback();
+    }
+
+    @FunctionalInterface
+    private interface SqlWork {
+        void execute(Connection connection) throws SQLException;
+    }
+
+    private static final class JdbcRuntimeException extends RuntimeException {
+        private JdbcRuntimeException(SQLException cause) {
+            super(cause);
+        }
+
+        @Override
+        public synchronized SQLException getCause() {
+            return (SQLException) super.getCause();
+        }
+    }
+}
--- a/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java
+++ b/modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfiguration.java
@@ -1,12 +1,18 @@
 package com.lingniu.ingest.tdenginehistory.config;

 import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema;
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryStatements;
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
+import com.lingniu.ingest.tdenginehistory.TdengineJdbcHistoryWriter;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;

+import javax.sql.DataSource;
+
@AutoConfiguration
@EnableConfigurationProperties(TdengineHistoryProperties.class)
 public class TdengineHistoryAutoConfiguration {
@@ -20,4 +26,25 @@ public class TdengineHistoryAutoConfiguration {
    public TdengineHistorySchema tdengineHistorySchema(TdengineHistoryProperties properties) {
        return new TdengineHistorySchema(properties.getDatabase());
    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    @ConditionalOnProperty(
+            prefix = "lingniu.ingest.tdengine-history",
+            name = "enabled",
+            havingValue = "true")
+    public TdengineHistoryStatements tdengineHistoryStatements(TdengineHistorySchema schema) {
+        return new TdengineHistoryStatements(schema);
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    @ConditionalOnBean(DataSource.class)
+    @ConditionalOnProperty(
+            prefix = "lingniu.ingest.tdengine-history",
+            name = "enabled",
+            havingValue = "true")
+    public TdengineHistoryWriter tdengineHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) {
+        return new TdengineJdbcHistoryWriter(dataSource, schema);
+    }
 }
--- 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
@@ -0,0 +1,208 @@
+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.Statement;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class TdengineJdbcHistoryWriterTest {
+
+    private final TdengineHistorySchema schema = new TdengineHistorySchema("vehicle_history");
+
+    @Test
+    void initializesSchemaWithBootstrapSql() throws Exception {
+        RecordingJdbc jdbc = new RecordingJdbc();
+        TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema);
+
+        writer.initializeSchema();
+
+        assertThat(jdbc.executedSql)
+                .containsExactlyElementsOf(schema.bootstrapSql());
+        assertThat(jdbc.commits).isEqualTo(1);
+    }
+
+    @Test
+    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"));
+
+        writer.appendRawFrames(List.of(first, second));
+
+        assertThat(jdbc.executedSql)
+                .filteredOn(sql -> sql.contains("USING raw_frames TAGS"))
+                .hasSize(1);
+        assertThat(jdbc.preparedBatches).hasSize(1);
+        RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next();
+        assertThat(batch.sql()).contains("INSERT INTO raw_jt808_");
+        assertThat(batch.rows()).hasSize(2);
+        assertThat(batch.rows().getFirst().get(1)).isEqualTo(Timestamp.from(first.ts()));
+        assertThat(batch.rows().getFirst().get(2)).isEqualTo("frame-1");
+        assertThat(batch.rows().get(1).get(2)).isEqualTo("frame-2");
+        assertThat(jdbc.commits).isEqualTo(1);
+    }
+
+    @Test
+    void writesLocationRowsWithNullableMileage() throws Exception {
+        RecordingJdbc jdbc = new RecordingJdbc();
+        TdengineJdbcHistoryWriter writer = new TdengineJdbcHistoryWriter(jdbc.dataSource(), schema);
+        TdengineLocationRow location = new TdengineLocationRow(
+                Instant.parse("2026-06-29T05:00:01Z"),
+                "fact-1",
+                "frame-1",
+                Instant.parse("2026-06-29T05:00:02Z"),
+                113.1,
+                23.4,
+                8.0,
+                42.0,
+                90.0,
+                1,
+                3,
+                null,
+                "archive://gb32960/frame-1.bin",
+                "{}",
+                "GB32960",
+                "vin:VIN123",
+                "VIN123",
+                "");
+
+        writer.appendLocations(List.of(location));
+
+        assertThat(jdbc.executedSql)
+                .anyMatch(sql -> sql.contains("USING vehicle_locations TAGS"));
+        RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next();
+        assertThat(batch.sql()).contains("INSERT INTO loc_gb32960_");
+        assertThat(batch.rows().getFirst().get(12)).isNull();
+    }
+
+    private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) {
+        return new TdengineRawFrameRow(
+                ts,
+                frameId,
+                ts.plusSeconds(1),
+                0x0200,
+                0,
+                ts,
+                "archive://jt808/" + frameId + ".bin",
+                "sha256:" + frameId,
+                67,
+                "SUCCEEDED",
+                "",
+                "10.0.0.1:808",
+                "{}",
+                "JT808",
+                "jt808:g7gps",
+                "VIN123",
+                "013800000000");
+    }
+
+    private static final class RecordingJdbc {
+        private final List<String> executedSql = new ArrayList<>();
+        private final Map<String, PreparedBatch> preparedBatches = new HashMap<>();
+        private int commits;
+        private int rollbacks;
+
+        DataSource dataSource() {
+            return (DataSource) Proxy.newProxyInstance(
+                    getClass().getClassLoader(),
+                    new Class<?>[]{DataSource.class},
+                    (proxy, method, args) -> switch (method.getName()) {
+                        case "getConnection" -> connection();
+                        case "unwrap" -> null;
+                        case "isWrapperFor" -> false;
+                        default -> defaultValue(method.getReturnType());
+                    });
+        }
+
+        private Connection connection() {
+            return (Connection) Proxy.newProxyInstance(
+                    getClass().getClassLoader(),
+                    new Class<?>[]{Connection.class},
+                    (proxy, method, args) -> switch (method.getName()) {
+                        case "createStatement" -> statement();
+                        case "prepareStatement" -> preparedStatement((String) args[0]);
+                        case "commit" -> {
+                            commits++;
+                            yield null;
+                        }
+                        case "rollback" -> {
+                            rollbacks++;
+                            yield null;
+                        }
+                        case "getAutoCommit" -> true;
+                        case "unwrap" -> null;
+                        case "isWrapperFor" -> false;
+                        default -> defaultValue(method.getReturnType());
+                    });
+        }
+
+        private Statement statement() {
+            return (Statement) Proxy.newProxyInstance(
+                    getClass().getClassLoader(),
+                    new Class<?>[]{Statement.class},
+                    (proxy, method, args) -> switch (method.getName()) {
+                        case "execute", "executeUpdate" -> {
+                            executedSql.add((String) args[0]);
+                            yield method.getReturnType() == boolean.class ? true : 1;
+                        }
+                        case "unwrap" -> null;
+                        case "isWrapperFor" -> false;
+                        default -> defaultValue(method.getReturnType());
+                    });
+        }
+
+        private PreparedStatement preparedStatement(String sql) {
+            PreparedBatch batch = preparedBatches.computeIfAbsent(sql, PreparedBatch::new);
+            Map<Integer, Object> current = new HashMap<>();
+            return (PreparedStatement) Proxy.newProxyInstance(
+                    getClass().getClassLoader(),
+                    new Class<?>[]{PreparedStatement.class},
+                    (proxy, method, args) -> switch (method.getName()) {
+                        case "setTimestamp", "setObject" -> {
+                            current.put((Integer) args[0], args[1]);
+                            yield null;
+                        }
+                        case "addBatch" -> {
+                            batch.rows().add(new HashMap<>(current));
+                            current.clear();
+                            yield null;
+                        }
+                        case "executeBatch" -> new int[batch.rows().size()];
+                        case "unwrap" -> null;
+                        case "isWrapperFor" -> false;
+                        default -> defaultValue(method.getReturnType());
+                    });
+        }
+
+        private static Object defaultValue(Class<?> type) {
+            if (type == boolean.class) {
+                return false;
+            }
+            if (type == int.class) {
+                return 0;
+            }
+            if (type == long.class) {
+                return 0L;
+            }
+            return null;
+        }
+
+        record PreparedBatch(String sql, List<Map<Integer, Object>> rows) {
+            PreparedBatch(String sql) {
+                this(sql, new ArrayList<>());
+            }
+        }
+    }
+}
--- a/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java
+++ b/modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/config/TdengineHistoryAutoConfigurationTest.java
@@ -1,11 +1,15 @@
 package com.lingniu.ingest.tdenginehistory.config;

 import com.lingniu.ingest.tdenginehistory.TdengineHistorySchema;
+import com.lingniu.ingest.tdenginehistory.TdengineHistoryWriter;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;

+import javax.sql.DataSource;
+
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;

 class TdengineHistoryAutoConfigurationTest {

@@ -32,4 +36,12 @@ class TdengineHistoryAutoConfigurationTest {
                            .isEqualTo("CREATE DATABASE IF NOT EXISTS vehicle_history_hot PRECISION 'ms'");
                });
    }
+
+    @Test
+    void createsWriterWhenEnabledAndDataSourceExists() {
+        contextRunner
+                .withBean(DataSource.class, () -> mock(DataSource.class))
+                .withPropertyValues("lingniu.ingest.tdengine-history.enabled=true")
+                .run(context -> assertThat(context).hasSingleBean(TdengineHistoryWriter.class));
+    }
 }