From 4de723d0b4a6d734c4121a20ffa52238d630a6f6 Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 13:28:36 +0800 Subject: [PATCH] feat: add tdengine jdbc history writer --- .../TdengineJdbcHistoryWriter.java | 147 +++++++++++++ .../TdengineHistoryAutoConfiguration.java | 27 +++ .../TdengineJdbcHistoryWriterTest.java | 208 ++++++++++++++++++ .../TdengineHistoryAutoConfigurationTest.java | 12 + 4 files changed, 394 insertions(+) create mode 100644 modules/sinks/tdengine-history-store/src/main/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriter.java create mode 100644 modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java 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 new file mode 100644 index 00000000..fa0d6e9c --- /dev/null +++ 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 rows) throws IOException { + append(rows, statements::rawFrame); + } + + @Override + public void appendLocations(List rows) throws IOException { + append(rows, statements::location); + } + + private void append(List rows, Function mapper) throws IOException { + if (rows == null || rows.isEmpty()) { + return; + } + inTransaction(connection -> { + Set createdChildTables = new LinkedHashSet<>(); + Map 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 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(); + } + } +} diff --git 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 index 4a7fc489..519a1156 100644 --- 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); + } } 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 new file mode 100644 index 00000000..1294353e --- /dev/null +++ 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 executedSql = new ArrayList<>(); + private final Map 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 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> rows) { + PreparedBatch(String sql) { + this(sql, new ArrayList<>()); + } + } + } +} diff --git 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 index 665abdde..b832c264 100644 --- 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)); + } }