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

fix: initialize tdengine schema before first write

Browse Source
This commit is contained in:
lingniu
2026-06-29 14:28:53 +08:00
parent 57ba7dca90
commit 2273d72e95
2 changed files with 42 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,6 +20,8 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter {
private final DataSource dataSource; private final DataSource dataSource;
private final TdengineHistorySchema schema; private final TdengineHistorySchema schema;
private final TdengineHistoryStatements statements; private final TdengineHistoryStatements statements;
private final Object schemaInitializationMonitor = new Object();
private volatile boolean schemaInitialized;
public TdengineJdbcHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) { public TdengineJdbcHistoryWriter(DataSource dataSource, TdengineHistorySchema schema) {
if (dataSource == null) { if (dataSource == null) {
@@ -41,6 +43,7 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter {
} }
} }
}); });
schemaInitialized = true;
} }
@Override @Override
@@ -62,7 +65,29 @@ public final class TdengineJdbcHistoryWriter implements TdengineHistoryWriter {
if (rows == null || rows.isEmpty()) { if (rows == null || rows.isEmpty()) {
return; return;
} }
if (!schemaInitialized) {
synchronized (schemaInitializationMonitor) {
if (!schemaInitialized) {
append(rows, mapper, true);
schemaInitialized = true;
return;
}
}
}
append(rows, mapper, false);
}
private <T> void append(List<T> rows,
Function<T, TdengineBatchStatement> mapper,
boolean initializeSchema) throws IOException {
inTransaction(connection -> { inTransaction(connection -> {
if (initializeSchema) {
try (Statement statement = connection.createStatement()) {
for (String sql : schema.bootstrapSql()) {
statement.execute(sql);
}
}
}
Set<String> createdChildTables = new LinkedHashSet<>(); Set<String> createdChildTables = new LinkedHashSet<>();
Map<String, PreparedStatement> preparedStatements = new LinkedHashMap<>(); Map<String, PreparedStatement> preparedStatements = new LinkedHashMap<>();
try { try {

17
modules/sinks/tdengine-history-store/src/test/java/com/lingniu/ingest/tdenginehistory/TdengineJdbcHistoryWriterTest.java
View File

@@ -41,6 +41,7 @@ class TdengineJdbcHistoryWriterTest {
writer.appendRawFrames(List.of(first, second)); writer.appendRawFrames(List.of(first, second));
assertThat(jdbc.executedSql).startsWith(schema.bootstrapSql().toArray(String[]::new));
assertThat(jdbc.executedSql) assertThat(jdbc.executedSql)
.filteredOn(sql -> sql.contains("USING raw_frames TAGS")) .filteredOn(sql -> sql.contains("USING raw_frames TAGS"))
.hasSize(1); .hasSize(1);
@@ -80,6 +81,7 @@ class TdengineJdbcHistoryWriterTest {
writer.appendLocations(List.of(location)); writer.appendLocations(List.of(location));
assertThat(jdbc.executedSql).startsWith(schema.bootstrapSql().toArray(String[]::new));
assertThat(jdbc.executedSql) assertThat(jdbc.executedSql)
.anyMatch(sql -> sql.contains("USING vehicle_locations TAGS")); .anyMatch(sql -> sql.contains("USING vehicle_locations TAGS"));
RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next(); RecordingJdbc.PreparedBatch batch = jdbc.preparedBatches.values().iterator().next();
@@ -96,6 +98,7 @@ class TdengineJdbcHistoryWriterTest {
writer.appendTelemetryFields(List.of(first, second)); writer.appendTelemetryFields(List.of(first, second));
assertThat(jdbc.executedSql).startsWith(schema.bootstrapSql().toArray(String[]::new));
assertThat(jdbc.executedSql) assertThat(jdbc.executedSql)
.filteredOn(sql -> sql.contains("USING telemetry_fields TAGS")) .filteredOn(sql -> sql.contains("USING telemetry_fields TAGS"))
.hasSize(1); .hasSize(1);
@@ -107,6 +110,20 @@ class TdengineJdbcHistoryWriterTest {
assertThat(batch.rows().get(1).get(6)).isEqualTo("124.5"); assertThat(batch.rows().get(1).get(6)).isEqualTo("124.5");
} }
@Test
void schemaBootstrapRunsOnlyOnceAcrossWrites() 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"))));
writer.appendRawFrames(List.of(rawFrame("frame-2", Instant.parse("2026-06-29T05:00:02Z"))));
assertThat(jdbc.executedSql)
.filteredOn(sql -> sql.startsWith("CREATE DATABASE"))
.hasSize(1);
assertThat(jdbc.commits).isEqualTo(2);
}
private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) { private static TdengineRawFrameRow rawFrame(String frameId, Instant ts) {
return new TdengineRawFrameRow( return new TdengineRawFrameRow(
ts, ts,