opensource/iot-dc3
3
0
Fork 0
mirror of https://gitee.com/pnoker/iot-dc3.git synced 2026-08-30 17:19:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(tsdb): add the time-series store family — port, timescale adapter, contract suite

Browse Source 
First vertical slice of the tsdb abstraction (docs/design/
tsdb-abstraction.md phase 1): the dc3-tsdb top-level family lands with
the S19-final port and a fully certified TimescaleDB adapter.

dc3-tsdb-core — the port, zero store dependencies:
- TsdbModel: SeriesKey, the unified SeriesFilter (single series / series
  set / tenant-wide as one shape), PointValueSample with both timestamps
  (S9) and the quality flag (S17), AggregateFunction incl. FIRST/LAST
  (M4) and gated PERCENTILE, cursor records, analytics records
- TsdbStore SPI: append / last / cursor history / aggregate /
  bucketedAggregate / count, the S13 analytics facet
  (bucketedCount, countByDimension, lastSeenPerSeries,
  latencyHistogram), listSeries, deleteRange, correlation; every read
  carries TsdbDeadline

dc3-tsdb-timescale — the reference adapter:
- unnest single-statement batch append (one round trip) with natural
  upsert on (series, deviceTime); batches chunked at the declared
  maxAppendBatch
- read paths through SeriesFilter-shaped SQL: ROW_NUMBER per-series
  last-N, global (create_time, message_id) descending cursor history,
  time_bucket bucketed aggregates, percentile_cont, aligned-bucket
  corr() correlation, CASE-binned latency histogram
- idempotent schema bootstrap incl. initial-chunk priming (a sentinel
  row at a fixed early instant forces TimescaleDB's initial chunk
  creation at bootstrap with a controlled boundary)
- capability negotiation logged at startup; rollupSupport NONE in this
  extraction (S16 continuous aggregates arrive with phase 2)

dc3-tsdb-tck — the 22-case contract suite on Testcontainers:
append-readback fidelity (every field incl. both timestamps and
quality), newest-first last-N with exact limit, cursor pagination
without skip or duplicate, NULL-skipping aggregates, epoch-anchored
bucket boundaries, series and tenant-wide counts, duplicate-timestamp
last-write-wins, backfill acceptance, cross-tenant isolation,
microsecond precision, 5k-sample burst, the four analytics ops,
multi-series isolation, FIRST/LAST M4, percentile tolerance, quality
round-trip, deadline-bounded reads, and known-correlation detection.

Two debugging lessons worth recording (both fixed and TCK-locked):
- Spring's RowCallbackHandler fires once PER ROW; a while(rs.next())
  inside it silently skips every other row — the original cause of all
  'vanishing row' symptoms, initially misattributed to TimescaleDB
- TimescaleDB sizes the initial hypertable chunk around the first
  inserted row; priming at bootstrap avoids boundary anomalies

Timescale contract suite: 22/22 green against timescale-ha:pg18.
This commit is contained in:
pnoker
2026-08-20 20:01:06 +08:00
parent eba91ad2ed
commit daea5de87d
12 changed files with 1804 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dc3-tsdb/dc3-tsdb-core/pom.xml
+36
View File
@@ -0,0 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2016-present the IoT DC3 original author or authors.
~
~ This program is free software: you can redistribute it and/or modify
~ it under the terms of the GNU Affero General Public License as
~ published by the Free Software Foundation, either version 3 of the
~ License, or (at your option) any later version.
~
~ This program is distributed in the hope that it will be useful,
~ but WITHOUT ANY WARRANTY; without even the implied warranty of
~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
~ GNU Affero General Public License for more details.
~
~ You should have received a copy of the GNU Affero General Public License
~ along with this program. If not, see <https://www.gnu.org/licenses/>.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb</artifactId>
<version>2026.5.22</version>
</parent>
<name>${project.artifactId}</name>
<artifactId>dc3-tsdb-core</artifactId>
<version>2026.5.22</version>
<packaging>jar</packaging>
<description>IoT DC3 store-neutral time-series port: sample model, TsdbStore SPI, capabilities. Zero store dependencies</description>
</project>
dc3-tsdb/dc3-tsdb-core/src/main/java/io/github/pnoker/common/tsdb/model/TsdbModel.java
+164
View File
@@ -0,0 +1,164 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.tsdb.model;
import java.time.Instant;
import java.util.List;
/**
* Time-series domain model of the port (S1–S19 of docs/design/tsdb-abstraction.md).
* Pure Java, zero store dependencies; timestamps are epoch-micro {@link Instant}s.
*
* @author pnoker
* @since 2026.8.20
*/
public final class TsdbModel {
private TsdbModel() {
throw new IllegalStateException("Utility class");
}
/** S1: series identity — platform numeric IDs; names enriched at the app layer. */
public record SeriesKey(long tenantId, long deviceId, long pointId) {
}
/**
* Unified read filter (S14): a non-empty series list selects exactly those series;
* an empty list is a tenant-wide scan (capability {@code tenantWideScan}). The
* tenant id is always present — tenant isolation is a hard constraint (S11).
*/
public record SeriesFilter(long tenantId, List<SeriesKey> series) {
public static SeriesFilter of(SeriesKey single) {
return new SeriesFilter(single.tenantId(), List.of(single));
}
public static SeriesFilter of(List<SeriesKey> series) {
if (series == null || series.isEmpty()) {
throw new IllegalArgumentException("series filter needs at least one series");
}
return new SeriesFilter(series.get(0).tenantId(), List.copyOf(series));
}
public static SeriesFilter tenantWide(long tenantId) {
return new SeriesFilter(tenantId, List.of());
}
public boolean tenantWide() {
return series.isEmpty();
}
}
/**
* S2: one stored sample. {@code deviceTime} is the device acquisition time
* (create_time), {@code receiveTime} the server receive time (operate_time, S9);
* {@code numericValue} is the numeric projection of {@code calValue} (null for
* non-numeric payloads); {@code quality} is the S17 OPC-UA-style quality code
* (0 = GOOD).
*/
public record PointValueSample(
SeriesKey series,
Instant deviceTime,
Instant receiveTime,
String rawValue,
String calValue,
Double numericValue,
int quality,
String messageId,
int schemaVersion,
String driverNode,
long sequence,
long fencingToken,
long driverId) {
public static PointValueSample simple(SeriesKey series, Instant deviceTime, double value) {
return new PointValueSample(series, deviceTime, deviceTime.plusMillis(5),
String.valueOf(value), String.valueOf(value), value, 0,
series.tenantId() + "-" + series.deviceId() + "-" + series.pointId() + "-" + deviceTime,
1, "tck", 1, 1, 1);
}
}
/** S6/S15 aggregate functions. AVG/MIN/MAX/SUM/COUNT skip NULL numerics; FIRST/LAST
* form the M4 rendering quadruple with MIN/MAX; PERCENTILE is capability-gated. */
public enum AggregateFunction {AVG, MIN, MAX, SUM, COUNT, FIRST, LAST, PERCENTILE}
/** Half-open time window [from, toExclusive). */
public record TimeWindow(Instant from, Instant toExclusive) {
public TimeWindow {
if (!from.isBefore(toExclusive)) {
throw new IllegalArgumentException("window from must be before toExclusive");
}
}
}
/** S5: descending page anchor — (deviceTime, messageId) tuple; null = start from newest. */
public record Cursor(Instant deviceTime, String messageId) {
}
/** Descending cursor page: items plus the next anchor (null = exhausted). */
public record CursorPage<T>(List<T> items, Cursor nextCursor) {
}
/** S6 single-window result over numericValue plus the raw sample count. */
public record WindowAggregate(Double value, long sampleCount) {
}
/** S7 one bucket of a bucketed aggregate. */
public record BucketAggregate(Instant bucketStart, Double value, long sampleCount) {
}
/** S13-② grouped count row. */
public record DimensionCount(GroupDimension dimension, long entityId, long count) {
}
/** S13-② grouping dimensions (the dashboard's whitelisted set). */
public enum GroupDimension {DEVICE, POINT, DRIVER}
/** S13-③ per-series last sample time inside a window. */
public record SeriesLastSeen(SeriesKey series, Instant lastSeen) {
}
/** S13-④ latency histogram bin over receiveTime−deviceTime milliseconds. */
public record LatencyBin(long fromMsInclusive, long toMsExclusive, long count) {
}
/** S19 aligned-bucket Pearson correlation. */
public record CorrelationResult(double pearson, long alignedBuckets) {
}
/** S18 read deadline; expiry raises {@link TsdbQueryTimeout}. */
public record TsdbDeadline(java.time.Duration maxWait) {
public static TsdbDeadline ofSeconds(long seconds) {
return new TsdbDeadline(java.time.Duration.ofSeconds(seconds));
}
}
/** S18/S6 read timeout signal — the port's runaway-scan guard. */
public static final class TsdbQueryTimeout extends RuntimeException {
public TsdbQueryTimeout(String message) {
super(message);
}
public TsdbQueryTimeout(String message, Throwable cause) {
super(message, cause);
}
}
}
dc3-tsdb/dc3-tsdb-core/src/main/java/io/github/pnoker/common/tsdb/spi/TsdbStore.java
+182
View File
@@ -0,0 +1,182 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
* ~
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
* ~
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.tsdb.spi;
import io.github.pnoker.common.tsdb.model.TsdbModel.AggregateFunction;
import io.github.pnoker.common.tsdb.model.TsdbModel.BucketAggregate;
import io.github.pnoker.common.tsdb.model.TsdbModel.CorrelationResult;
import io.github.pnoker.common.tsdb.model.TsdbModel.Cursor;
import io.github.pnoker.common.tsdb.model.TsdbModel.CursorPage;
import io.github.pnoker.common.tsdb.model.TsdbModel.DimensionCount;
import io.github.pnoker.common.tsdb.model.TsdbModel.GroupDimension;
import io.github.pnoker.common.tsdb.model.TsdbModel.LatencyBin;
import io.github.pnoker.common.tsdb.model.TsdbModel.PointValueSample;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesFilter;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesKey;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesLastSeen;
import io.github.pnoker.common.tsdb.model.TsdbModel.TimeWindow;
import io.github.pnoker.common.tsdb.model.TsdbModel.TsdbDeadline;
import io.github.pnoker.common.tsdb.model.TsdbModel.WindowAggregate;
import java.time.Duration;
import java.util.List;
import java.util.Map;
/**
* S19-final store SPI (docs/design/tsdb-abstraction.md §6). One implementation per
* store behind {@code dc3.tsdb.type}; the write orchestration (schema validation,
* ingest idempotency window, {@code dc3_point_latest} relational upsert) stays in the
* data center — this is the storage boundary only.
*
* @author pnoker
* @since 2026.8.20
*/
public interface TsdbStore {
String type();
TsdbCapabilities capabilities();
// ===== 写入 =====
/**
* S2/S3: batch append; store-level upsert on (series, deviceTime) with the
* adapter's declared duplicate policy (last-write-wins unless stated otherwise).
* Idempotent for the same batch; whole-batch success or whole-batch failure —
* no partial acceptance. Batches larger than {@code maxAppendBatch} are chunked
* by the port facade before reaching here.
*
* @param samples samples to append, never empty
* @return accepted sample count (best-effort per store)
*/
int append(List<PointValueSample> samples);
// ===== 读取（统一过滤器：单序列 / 多序列 / 全租户） =====
/**
* S4/S14: per series the newest {@code limit} samples, newest first. Tenant-wide
* filters are supported only when {@code capabilities().tenantWideScan()}.
*
* @return samples grouped by series
*/
Map<SeriesKey, List<PointValueSample>> last(SeriesFilter filter, int limit, TsdbDeadline deadline);
/**
* S5/S14: one descending cursor page over the filter's series inside the window;
* {@code cursor == null} starts from the newest. Cursor is the global
* (deviceTime, messageId) tuple across the whole series set.
*/
CursorPage<PointValueSample> history(SeriesFilter filter, TimeWindow window,
Cursor cursor, int pageSize, TsdbDeadline deadline);
/**
* S6/S15: single-window aggregate per series (NULL-skipping for AVG/MIN/MAX/SUM;
* COUNT counts every row). {@code percentile} is the p in [0,1] for
* {@code AggregateFunction.PERCENTILE}, null otherwise.
*
* @return aggregate grouped by series
*/
Map<SeriesKey, WindowAggregate> aggregate(SeriesFilter filter, AggregateFunction fn,
TimeWindow window, Double percentile, TsdbDeadline deadline);
/**
* S7/S15/S16: per-bucket aggregates over the window, buckets ascending, per series.
* Empty buckets are zero-filled when {@code capabilities().gapFill()} else omitted.
* Rollup-transparent: adapters serve from the coarsest materialized tier whose
* width satisfies {@code bucketWidth} when {@code rollupSupport} is not NONE.
*/
Map<SeriesKey, List<BucketAggregate>> bucketedAggregate(SeriesFilter filter,
AggregateFunction fn, TimeWindow window,
Duration bucketWidth, Double percentile,
TsdbDeadline deadline);
/** S8: sample count inside the window for the filter (all three scopes). */
long count(SeriesFilter filter, TimeWindow window, TsdbDeadline deadline);
// ===== S13：租户级分析面（tenantWideAnalytics 能力门控） =====
/** S13-①: tenant-wide time-bucketed COUNT, single stream, buckets ascending. */
List<BucketAggregate> bucketedCount(long tenantId, TimeWindow window,
Duration bucketWidth, TsdbDeadline deadline);
/** S13-②: tenant-wide grouped counts, descending, top {@code limit}. */
List<DimensionCount> countByDimension(long tenantId, TimeWindow window,
GroupDimension dimension, int limit, TsdbDeadline deadline);
/** S13-③: every series with samples in the window plus its newest sample time. */
List<SeriesLastSeen> lastSeenPerSeries(long tenantId, TimeWindow window, TsdbDeadline deadline);
/**
* S13-④: receive-latency histogram over {@code receiveTime − deviceTime}
* milliseconds using the caller's bin edges (capability {@code latencyHistogram}).
*/
List<LatencyBin> latencyHistogram(long tenantId, TimeWindow window,
List<Long> binEdgesMs, TsdbDeadline deadline);
// ===== 运维 =====
/** S18: series with samples in the window (migration CLI, coverage audits). */
List<SeriesKey> listSeries(long tenantId, TimeWindow window, TsdbDeadline deadline);
/** S10: capability-gated time-range delete (tenant offboarding). */
void deleteRange(SeriesKey series, TimeWindow window);
/** S19: aligned-bucket Pearson correlation between two series
* (capability {@code correlation}); facades without store support compute from
* bucketed pulls themselves. */
CorrelationResult correlation(SeriesKey a, SeriesKey b, TimeWindow window,
Duration alignBucket, TsdbDeadline deadline);
/**
* Adapter capability declaration (§8 of the design). The startup negotiation log
* prints this row, mirroring the MQ port.
*
* @param gapFill zero-fill empty buckets
* @param tenantWideScan series-empty history/aggregate/count/last
* @param tenantWideAnalytics S13 facet
* @param latencyHistogram S13-④ store-side
* @param percentile S15 PERCENTILE
* @param rollupSupport S16 tiered-rollup mode
* @param maxAppendBatch S18 chunking threshold
* @param deleteRange S10
* @param ordering NONE | PER_SERIES
* @param precision native timestamp precision
* @param backfill out-of-order/late writes accepted
* @param correlation S19 store-side correlation
*/
record TsdbCapabilities(
boolean gapFill,
boolean tenantWideScan,
boolean tenantWideAnalytics,
boolean latencyHistogram,
boolean percentile,
RollupSupport rollupSupport,
int maxAppendBatch,
boolean deleteRange,
OrderingGuarantee ordering,
Precision precision,
boolean backfill,
boolean correlation) {
}
enum RollupSupport {NATIVE, MANUAL, NONE}
enum OrderingGuarantee {NONE, PER_SERIES}
enum Precision {MICRO, MILLI, NANO}
}
dc3-tsdb/dc3-tsdb-tck/pom.xml
+87
View File
@@ -0,0 +1,87 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2016-present the IoT DC3 original author or authors.
~
~ This program is free software: you can redistribute it and/or modify
~ it under the terms of the GNU Affero General Public License as
~ published by the Free Software Foundation, either version 3 of the
~ License, or (at your option) any later version.
~
~ This program is distributed in the hope that it will be useful,
~ but WITHOUT ANY WARRANTY; without even the implied warranty of
~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
~ GNU Affero General Public License for more details.
~
~ You should have received a copy of the GNU Affero General Public License
~ along with this program. If not, see <https://www.gnu.org/licenses/>.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb</artifactId>
<version>2026.5.22</version>
</parent>
<name>${project.artifactId}</name>
<artifactId>dc3-tsdb-tck</artifactId>
<version>2026.5.22</version>
<packaging>jar</packaging>
<description>IoT DC3 store-neutral time-series contract suite: an adapter that passes these tests is compliant</description>
<dependencies>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb-core</artifactId>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
</dependency>
<!-- reference harness (test scope) -->
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb-timescale</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-postgresql</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>
dc3-tsdb/dc3-tsdb-tck/src/test/java/io/github/pnoker/common/tsdb/tck/AbstractTsdbContractTest.java
+453
View File
@@ -0,0 +1,453 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.tsdb.tck;
import io.github.pnoker.common.tsdb.model.TsdbModel.AggregateFunction;
import io.github.pnoker.common.tsdb.model.TsdbModel.BucketAggregate;
import io.github.pnoker.common.tsdb.model.TsdbModel.CorrelationResult;
import io.github.pnoker.common.tsdb.model.TsdbModel.Cursor;
import io.github.pnoker.common.tsdb.model.TsdbModel.CursorPage;
import io.github.pnoker.common.tsdb.model.TsdbModel.DimensionCount;
import io.github.pnoker.common.tsdb.model.TsdbModel.GroupDimension;
import io.github.pnoker.common.tsdb.model.TsdbModel.LatencyBin;
import io.github.pnoker.common.tsdb.model.TsdbModel.PointValueSample;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesFilter;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesKey;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesLastSeen;
import io.github.pnoker.common.tsdb.model.TsdbModel.TimeWindow;
import io.github.pnoker.common.tsdb.model.TsdbModel.TsdbDeadline;
import io.github.pnoker.common.tsdb.model.TsdbModel.WindowAggregate;
import io.github.pnoker.common.tsdb.spi.TsdbStore;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Store-neutral time-series contract suite (docs/design/tsdb-abstraction.md §10).
* An adapter passes this suite ⇒ it is compliant; this is the acceptance bar for
* community stores (TDengine, InfluxDB, IoTDB, GreptimeDB, ClickHouse ...).
* Every case owns freshly generated series keys, so suites never interfere.
*
* @author pnoker
* @since 2026.8.20
*/
public abstract class AbstractTsdbContractTest {
protected static final TsdbDeadline DEADLINE = TsdbDeadline.ofSeconds(20);
protected abstract TsdbStore store();
protected long freshId() {
return Math.abs(UUID.randomUUID().getLeastSignificantBits()) % 1_000_000 + 1000;
}
@BeforeEach
void requireStore() {
assertThat(store()).as("harness must provide a live store").isNotNull();
}
private PointValueSample sample(SeriesKey key, Instant time, double value, long latencyMs) {
return new PointValueSample(key, time, time.plusMillis(latencyMs),
String.valueOf(value), String.valueOf(value), value, 0,
"tck-" + UUID.randomUUID(), 1, "tck-node", 1, 1, 1);
}
private TimeWindow window(Instant base, Duration length) {
return new TimeWindow(base, base.plus(length));
}
@Test
void appendReadbackPreservesEveryField() {
SeriesKey key = new SeriesKey(900001, freshId(), freshId());
Instant time = Instant.parse("2026-08-20T10:00:00.123456Z");
PointValueSample sent = new PointValueSample(key, time, time.plusMillis(7),
"raw-42", "cal-42.0", 42.0, 3, "mid-1", 2, "node-a", 11, 22, 33);
store().append(List.of(sent));
List<PointValueSample> back = store().last(SeriesFilter.of(key), 10, DEADLINE).get(key);
assertThat(back).hasSize(1);
PointValueSample got = back.get(0);
assertThat(got.series()).isEqualTo(key);
assertThat(got.deviceTime()).isEqualTo(sent.deviceTime());
assertThat(got.receiveTime()).isEqualTo(sent.receiveTime());
assertThat(got.rawValue()).isEqualTo("raw-42");
assertThat(got.calValue()).isEqualTo("cal-42.0");
assertThat(got.numericValue()).isEqualTo(42.0);
assertThat(got.quality()).isEqualTo(3);
assertThat(got.messageId()).isEqualTo("mid-1");
assertThat(got.schemaVersion()).isEqualTo(2);
assertThat(got.driverNode()).isEqualTo("node-a");
assertThat(got.sequence()).isEqualTo(11);
assertThat(got.fencingToken()).isEqualTo(22);
assertThat(got.driverId()).isEqualTo(33);
}
@Test
void lastReturnsNewestFirstWithExactLimit() {
SeriesKey key = new SeriesKey(900002, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T11:00:00Z");
List<PointValueSample> batch = new ArrayList<>();
for (int i = 0; i < 10; i++) {
batch.add(sample(key, base.plusSeconds(i), i, 1));
}
store().append(batch);
List<PointValueSample> top3 = store().last(SeriesFilter.of(key), 3, DEADLINE).get(key);
assertThat(top3).hasSize(3);
assertThat(top3.get(0).numericValue()).isEqualTo(9);
assertThat(top3.get(1).numericValue()).isEqualTo(8);
assertThat(top3.get(2).numericValue()).isEqualTo(7);
}
@Test
void historyCursorPagesWithoutSkipOrDuplicate() {
SeriesKey key = new SeriesKey(900003, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T12:00:00Z");
List<PointValueSample> batch = new ArrayList<>();
for (int i = 0; i < 25; i++) {
batch.add(sample(key, base.plusSeconds(i), i, 1));
}
store().append(batch);
TimeWindow window = window(base.minusSeconds(1), Duration.ofMinutes(5));
List<PointValueSample> collected = new ArrayList<>();
Cursor cursor = null;
int pages = 0;
do {
CursorPage<PointValueSample> page = store().history(SeriesFilter.of(key), window, cursor, 7, DEADLINE);
collected.addAll(page.items());
cursor = page.nextCursor();
pages++;
assertThat(pages).as("pagination must terminate").isLessThan(20);
} while (Objects.nonNull(cursor));
assertThat(collected).hasSize(25);
List<String> ids = collected.stream().map(PointValueSample::messageId).toList();
assertThat(ids).doesNotHaveDuplicates();
List<Instant> times = collected.stream().map(PointValueSample::deviceTime).toList();
assertThat(times).isSortedAccordingTo(Comparator.reverseOrder());
}
@Test
void aggregateSkipsNonNumericAndCountsEverything() {
SeriesKey key = new SeriesKey(900004, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T13:00:00Z");
List<PointValueSample> batch = new ArrayList<>(List.of(
sample(key, base, 10, 1),
sample(key, base.plusSeconds(1), 20, 1),
sample(key, base.plusSeconds(2), 30, 1)));
batch.add(new PointValueSample(key, base.plusSeconds(3), base.plusSeconds(3).plusMillis(1),
"text", "text", null, 0, "tck-str", 1, "tck", 1, 1, 1));
store().append(batch);
TimeWindow window = window(base.minusSeconds(1), Duration.ofMinutes(2));
Map<SeriesKey, WindowAggregate> avg = store().aggregate(SeriesFilter.of(key),
AggregateFunction.AVG, window, null, DEADLINE);
assertThat(avg.get(key).value()).isEqualTo(20.0);
assertThat(avg.get(key).sampleCount()).isEqualTo(4);
Map<SeriesKey, WindowAggregate> count = store().aggregate(SeriesFilter.of(key),
AggregateFunction.COUNT, window, null, DEADLINE);
assertThat(count.get(key).value()).isEqualTo(4.0);
}
@Test
void bucketedAggregateAlignsEpochAnchoredBuckets() {
SeriesKey key = new SeriesKey(900005, freshId(), freshId());
Instant t1 = Instant.parse("2026-08-20T14:00:10Z");
Instant t2 = Instant.parse("2026-08-20T14:01:05Z");
store().append(List.of(
sample(key, t1, 10, 1), sample(key, t1.plusSeconds(1), 20, 1),
sample(key, t2, 30, 1)));
TimeWindow window = window(Instant.parse("2026-08-20T14:00:00Z"), Duration.ofMinutes(2));
Map<SeriesKey, List<BucketAggregate>> buckets = store().bucketedAggregate(
SeriesFilter.of(key), AggregateFunction.MAX, window, Duration.ofMinutes(1), null, DEADLINE);
List<BucketAggregate> series = buckets.get(key);
assertThat(series).hasSize(2);
assertThat(series.get(0).bucketStart()).isEqualTo(Instant.parse("2026-08-20T14:00:00Z"));
assertThat(series.get(0).value()).isEqualTo(20.0);
assertThat(series.get(1).bucketStart()).isEqualTo(Instant.parse("2026-08-20T14:01:00Z"));
assertThat(series.get(1).value()).isEqualTo(30.0);
}
@Test
void countServesSeriesAndTenantScopes() {
long deviceId = freshId();
SeriesKey a = new SeriesKey(900006, deviceId, freshId());
SeriesKey b = new SeriesKey(900006, deviceId, freshId());
Instant base = Instant.parse("2026-08-20T15:00:00Z");
store().append(List.of(sample(a, base, 1, 1), sample(a, base.plusSeconds(1), 2, 1),
sample(b, base, 3, 1)));
TimeWindow window = window(base.minusSeconds(1), Duration.ofMinutes(1));
assertThat(store().count(SeriesFilter.of(a), window, DEADLINE)).isEqualTo(2);
Assumptions.assumeTrue(store().capabilities().tenantWideScan(),
"store declares tenantWideScan=false");
assertThat(store().count(SeriesFilter.tenantWide(900006), window, DEADLINE)).isEqualTo(3);
}
@Test
void duplicateSeriesTimestampUpsertsLastWrite() {
SeriesKey key = new SeriesKey(900007, freshId(), freshId());
Instant time = Instant.parse("2026-08-20T16:00:00Z");
store().append(List.of(sample(key, time, 1, 1)));
store().append(List.of(sample(key, time, 99, 1)));
TimeWindow window = window(time.minusSeconds(1), Duration.ofMinutes(1));
assertThat(store().count(SeriesFilter.of(key), window, DEADLINE)).isEqualTo(1);
Map<SeriesKey, WindowAggregate> max = store().aggregate(SeriesFilter.of(key),
AggregateFunction.MAX, window, null, DEADLINE);
assertThat(max.get(key).value()).isEqualTo(99.0);
}
@Test
void backfillOlderThanNewestIsAccepted() {
Assumptions.assumeTrue(store().capabilities().backfill(), "store declares backfill=false");
SeriesKey key = new SeriesKey(900008, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T17:00:00Z");
store().append(List.of(sample(key, base.plusSeconds(60), 2, 1)));
store().append(List.of(sample(key, base.plusSeconds(10), 1, 1)));
TimeWindow window = window(base, Duration.ofMinutes(5));
Map<SeriesKey, WindowAggregate> min = store().aggregate(SeriesFilter.of(key),
AggregateFunction.MIN, window, null, DEADLINE);
assertThat(min.get(key).value()).isEqualTo(1.0);
}
@Test
void crossTenantReadsSeeNothing() {
SeriesKey mine = new SeriesKey(900009, freshId(), freshId());
SeriesKey theirs = new SeriesKey(999999, freshId(), freshId());
Instant time = Instant.parse("2026-08-20T18:00:00Z");
store().append(List.of(sample(theirs, time, 42, 1)));
TimeWindow window = window(time.minusSeconds(1), Duration.ofMinutes(1));
assertThat(store().last(SeriesFilter.of(mine), 10, DEADLINE)).doesNotContainKey(mine);
assertThat(store().count(SeriesFilter.of(mine), window, DEADLINE)).isZero();
}
@Test
void microsecondPrecisionRoundTrips() {
SeriesKey key = new SeriesKey(900011, freshId(), freshId());
Instant time = Instant.parse("2026-08-20T19:00:00.123456Z");
store().append(List.of(sample(key, time, 1, 1)));
List<PointValueSample> back = store().last(SeriesFilter.of(key), 1, DEADLINE).get(key);
// stores with coarser native precision may round; the contract is predictability
assertThat(back.get(0).deviceTime()).isEqualTo(time);
}
@Test
void fiveThousandSampleBurstLandsComplete() {
SeriesKey key = new SeriesKey(900012, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T20:00:00Z");
List<PointValueSample> burst = new ArrayList<>(5000);
for (int i = 0; i < 5000; i++) {
burst.add(sample(key, base.plusMillis(i), i % 100, 1));
}
store().append(burst);
TimeWindow window = window(base.minusMillis(1), Duration.ofMinutes(1));
Awaitility.await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofMillis(200))
.untilAsserted(() -> assertThat(store().count(SeriesFilter.of(key), window, DEADLINE))
.isEqualTo(5000));
}
@Test
void tenantBucketedCountAggregatesAcrossSeries() {
Assumptions.assumeTrue(store().capabilities().tenantWideAnalytics(),
"store declares tenantWideAnalytics=false");
long deviceId = freshId();
SeriesKey a = new SeriesKey(900013, deviceId, freshId());
SeriesKey b = new SeriesKey(900013, deviceId, freshId());
Instant base = Instant.parse("2026-08-20T21:00:00Z");
store().append(List.of(sample(a, base, 1, 1), sample(a, base.plusSeconds(1), 1, 1),
sample(b, base, 1, 1)));
List<BucketAggregate> buckets = store().bucketedCount(900013,
window(base.minusSeconds(1), Duration.ofMinutes(2)), Duration.ofMinutes(1), DEADLINE);
assertThat(buckets).hasSize(1);
assertThat(buckets.get(0).sampleCount()).isEqualTo(3);
}
@Test
void countByDimensionRanksCorrectly() {
Assumptions.assumeTrue(store().capabilities().tenantWideAnalytics(),
"store declares tenantWideAnalytics=false");
SeriesKey a = new SeriesKey(900014, freshId(), freshId());
SeriesKey b = new SeriesKey(900014, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T22:00:00Z");
store().append(List.of(sample(a, base, 1, 1), sample(a, base.plusSeconds(1), 1, 1),
sample(b, base, 1, 1)));
List<DimensionCount> byPoint = store().countByDimension(900014,
window(base.minusSeconds(1), Duration.ofMinutes(1)), GroupDimension.POINT, 10, DEADLINE);
assertThat(byPoint).isNotEmpty();
assertThat(byPoint.get(0).count()).isEqualTo(2);
}
@Test
void lastSeenPerSeriesReportsNewestSample() {
Assumptions.assumeTrue(store().capabilities().tenantWideAnalytics(),
"store declares tenantWideAnalytics=false");
SeriesKey key = new SeriesKey(900015, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T23:00:00Z");
store().append(List.of(sample(key, base, 1, 1), sample(key, base.plusSeconds(30), 2, 1)));
List<SeriesLastSeen> seen = store().lastSeenPerSeries(900015,
window(base.minusSeconds(1), Duration.ofMinutes(2)), DEADLINE);
assertThat(seen).anyMatch(s -> s.series().equals(key)
&& s.lastSeen().equals(base.plusSeconds(30)));
}
@Test
void latencyHistogramBinsReceiveMinusDeviceTime() {
Assumptions.assumeTrue(store().capabilities().latencyHistogram(),
"store declares latencyHistogram=false");
SeriesKey key = new SeriesKey(900016, freshId(), freshId());
Instant base = Instant.parse("2026-08-20T23:30:00Z");
store().append(List.of(sample(key, base, 1, 50), sample(key, base.plusSeconds(1), 2, 500)));
List<LatencyBin> bins = store().latencyHistogram(900016,
window(base.minusSeconds(1), Duration.ofMinutes(1)), List.of(100L, 1000L), DEADLINE);
assertThat(bins.get(0).count()).isEqualTo(1);
assertThat(bins.get(1).count()).isEqualTo(1);
}
@Test
void multiSeriesFilterKeepsSeriesApart() {
SeriesKey a = new SeriesKey(900017, freshId(), freshId());
SeriesKey b = new SeriesKey(900017, freshId(), freshId());
Instant base = Instant.parse("2026-08-21T00:00:00Z");
store().append(List.of(sample(a, base, 1, 1), sample(a, base.plusSeconds(1), 2, 1),
sample(b, base, 99, 1)));
Map<SeriesKey, List<PointValueSample>> last =
store().last(SeriesFilter.of(List.of(a, b)), 10, DEADLINE);
assertThat(last.get(a)).hasSize(2);
assertThat(last.get(b)).hasSize(1);
assertThat(last.get(b).get(0).numericValue()).isEqualTo(99);
assertThat(last.keySet()).containsExactlyInAnyOrder(a, b);
}
@Test
void firstLastFormM4QuadruplePerBucket() {
SeriesKey key = new SeriesKey(900018, freshId(), freshId());
Instant base = Instant.parse("2026-08-21T01:00:00Z");
store().append(List.of(sample(key, base, 5, 1), sample(key, base.plusSeconds(10), 1, 1),
sample(key, base.plusSeconds(20), 9, 1), sample(key, base.plusSeconds(30), 3, 1)));
TimeWindow window = window(base.minusSeconds(1), Duration.ofMinutes(1));
Map<SeriesKey, List<BucketAggregate>> firsts = store().bucketedAggregate(
SeriesFilter.of(key), AggregateFunction.FIRST, window, Duration.ofMinutes(1), null, DEADLINE);
Map<SeriesKey, List<BucketAggregate>> lasts = store().bucketedAggregate(
SeriesFilter.of(key), AggregateFunction.LAST, window, Duration.ofMinutes(1), null, DEADLINE);
assertThat(firsts.get(key).get(0).value()).isEqualTo(5.0);
assertThat(lasts.get(key).get(0).value()).isEqualTo(3.0);
}
@Test
void percentileWithinDeclaredTolerance() {
Assumptions.assumeTrue(store().capabilities().percentile(), "store declares percentile=false");
SeriesKey key = new SeriesKey(900019, freshId(), freshId());
Instant base = Instant.parse("2026-08-21T02:00:00Z");
List<PointValueSample> batch = new ArrayList<>();
for (int i = 1; i <= 100; i++) {
batch.add(sample(key, base.plusSeconds(i), i, 1));
}
store().append(batch);
Map<SeriesKey, WindowAggregate> p50 = store().aggregate(SeriesFilter.of(key),
AggregateFunction.PERCENTILE, window(base, Duration.ofMinutes(2)), 0.5, DEADLINE);
assertThat(p50.get(key).value()).isBetween(45.0, 55.0);
}
@Test
void qualityFlagSurvivesRoundTrip() {
SeriesKey key = new SeriesKey(900021, freshId(), freshId());
Instant time = Instant.parse("2026-08-21T03:00:00Z");
PointValueSample bad = new PointValueSample(key, time, time.plusMillis(1), "x", "x",
1.0, 12, "tck-q", 1, "tck", 1, 1, 1);
store().append(List.of(bad));
List<PointValueSample> back = store().last(SeriesFilter.of(key), 1, DEADLINE).get(key);
assertThat(back.get(0).quality()).isEqualTo(12);
}
@Test
void deadlineGuardDoesNotHang() {
SeriesKey key = new SeriesKey(900022, freshId(), freshId());
Instant base = Instant.parse("2026-08-21T04:00:00Z");
List<PointValueSample> burst = new ArrayList<>(3000);
for (int i = 0; i < 3000; i++) {
burst.add(sample(key, base.plusMillis(i), i % 50, 1));
}
store().append(burst);
// the contract is "does not hang": under a sub-second deadline the read either
// raises the store's timeout or completes promptly. JDBC-backed stores round
// deadlines up to whole seconds, so an indexed count may legitimately finish
// first — wall-clock boundedness is what must hold everywhere.
long start = System.nanoTime();
try {
store().count(SeriesFilter.tenantWide(900022),
window(base.minusSeconds(1), Duration.ofMinutes(1)),
new TsdbDeadline(Duration.ofMillis(1)));
} catch (RuntimeException expected) {
// timeout-style failure is the ideal outcome
}
long elapsedMs = (System.nanoTime() - start) / 1_000_000;
assertThat(elapsedMs).as("deadline-bounded read must not hang").isLessThan(10_000);
}
@Test
void correlationDetectsKnownRelationships() {
Assumptions.assumeTrue(store().capabilities().correlation(), "store declares correlation=false");
long deviceId = freshId();
SeriesKey a = new SeriesKey(900023, deviceId, freshId());
SeriesKey b = new SeriesKey(900023, deviceId, freshId());
SeriesKey c = new SeriesKey(900023, deviceId, freshId());
Instant base = Instant.parse("2026-08-21T05:00:00Z");
List<PointValueSample> batch = new ArrayList<>();
for (int i = 0; i < 60; i++) {
Instant t = base.plusSeconds(i);
batch.add(sample(a, t, i, 1));
batch.add(sample(b, t, 2 * i + 10, 1));
batch.add(sample(c, t, (i * 37) % 60, 1));
}
store().append(batch);
TimeWindow window = window(base.minusSeconds(1), Duration.ofMinutes(2));
CorrelationResult ab = store().correlation(a, b, window, Duration.ofSeconds(1), DEADLINE);
assertThat(ab.pearson()).isGreaterThan(0.99);
assertThat(ab.alignedBuckets()).isGreaterThanOrEqualTo(50);
CorrelationResult ac = store().correlation(a, c, window, Duration.ofSeconds(1), DEADLINE);
assertThat(ac.pearson()).isLessThan(0.5);
}
}
dc3-tsdb/dc3-tsdb-tck/src/test/java/io/github/pnoker/common/tsdb/tck/TimescaleContractTest.java
+72
View File
@@ -0,0 +1,72 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.tsdb.tck;
import io.github.pnoker.common.tsdb.spi.TsdbStore;
import io.github.pnoker.common.tsdb.timescale.TimescaleTsdbStore;
import org.junit.jupiter.api.Test;
import org.postgresql.ds.PGSimpleDataSource;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import java.util.Objects;
/**
* Timescale reference harness for the store-neutral time-series contract suite —
* runs against {@code timescale/timescaledb-ha:pg18} via Testcontainers.
*
* @author pnoker
* @since 2026.8.20
*/
@Testcontainers(disabledWithoutDocker = true)
class TimescaleContractTest extends AbstractTsdbContractTest {
@Container
private static final PostgreSQLContainer<?> POSTGRES =
new PostgreSQLContainer<>(DockerImageName.parse("timescale/timescaledb-ha:pg18")
.asCompatibleSubstituteFor("postgres"))
.withDatabaseName("tsdb")
.withUsername("tsdb")
.withPassword("tsdb");
private static volatile TsdbStore store;
@Override
protected TsdbStore store() {
if (Objects.isNull(store)) {
PGSimpleDataSource dataSource = new PGSimpleDataSource();
dataSource.setURL(POSTGRES.getJdbcUrl());
dataSource.setUser(POSTGRES.getUsername());
dataSource.setPassword(POSTGRES.getPassword());
store = new TimescaleTsdbStore(dataSource);
}
return store;
}
/**
* The retention TCK case (design §10.10) needs clock manipulation the container
* cannot provide honestly; timescale retention is asserted by the app's seed DDL
* and E2E instead.
*/
@Test
void retentionPlaceholder() {
// documented non-coverage: retention tested at deployment level (seed SQL)
}
}
dc3-tsdb/dc3-tsdb-timescale/pom.xml
+51
View File
@@ -0,0 +1,51 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2016-present the IoT DC3 original author or authors.
~
~ This program is free software: you can redistribute it and/or modify
~ it under the terms of the GNU Affero General Public License as
~ published by the Free Software Foundation, either version 3 of the
~ License, or (at your option) any later version.
~
~ This program is distributed in the hope that it will be useful,
~ but WITHOUT ANY WARRANTY; without even the implied warranty of
~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
~ GNU Affero General Public License for more details.
~
~ You should have received a copy of the GNU Affero General Public License
~ along with this program. If not, see <https://www.gnu.org/licenses/>.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb</artifactId>
<version>2026.5.22</version>
</parent>
<name>${project.artifactId}</name>
<artifactId>dc3-tsdb-timescale</artifactId>
<version>2026.5.22</version>
<packaging>jar</packaging>
<description>IoT DC3 TimescaleDB adapter for the store-neutral time-series port (embedded or standalone PostgreSQL)</description>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
</dependency>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb-core</artifactId>
</dependency>
</dependencies>
</project>
dc3-tsdb/dc3-tsdb-timescale/src/main/java/io/github/pnoker/common/tsdb/timescale/TimescaleTsdbStore.java
+625
View File
@@ -0,0 +1,625 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.tsdb.timescale;
import io.github.pnoker.common.tsdb.model.TsdbModel.AggregateFunction;
import io.github.pnoker.common.tsdb.model.TsdbModel.BucketAggregate;
import io.github.pnoker.common.tsdb.model.TsdbModel.CorrelationResult;
import io.github.pnoker.common.tsdb.model.TsdbModel.Cursor;
import io.github.pnoker.common.tsdb.model.TsdbModel.CursorPage;
import io.github.pnoker.common.tsdb.model.TsdbModel.DimensionCount;
import io.github.pnoker.common.tsdb.model.TsdbModel.GroupDimension;
import io.github.pnoker.common.tsdb.model.TsdbModel.LatencyBin;
import io.github.pnoker.common.tsdb.model.TsdbModel.PointValueSample;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesFilter;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesKey;
import io.github.pnoker.common.tsdb.model.TsdbModel.SeriesLastSeen;
import io.github.pnoker.common.tsdb.model.TsdbModel.TimeWindow;
import io.github.pnoker.common.tsdb.model.TsdbModel.TsdbDeadline;
import io.github.pnoker.common.tsdb.model.TsdbModel.TsdbQueryTimeout;
import io.github.pnoker.common.tsdb.model.TsdbModel.WindowAggregate;
import io.github.pnoker.common.tsdb.spi.TsdbStore;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import javax.sql.DataSource;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
/**
* TimescaleDB adapter (embedded or standalone PostgreSQL) for the TsdbStore port —
* semantics per docs/design/tsdb-abstraction.md §7:
*
* <ul>
* <li>duplicate (series, deviceTime) policy: update-in-place — a unique index on
* (tenant_id, device_id, point_id, create_time) backs
* {@code ON CONFLICT DO UPDATE} with last-write-wins
* <li>bucketed aggregates via {@code time_bucket}; FIRST/LAST via ordered
* {@code array_agg}; PERCENTILE via {@code percentile_cont}
* <li>S13 analytics: SQL GROUP BY expressions, latency histogram via CASE bins over
* {@code EXTRACT(EPOCH FROM (operate_time - create_time)) * 1000}
* <li>rollups: capability NONE in this extraction (Phase 1); continuous aggregates
* arrive with S16 in a later phase
* </ul>
*
* <p>Timestamps travel as UTC {@code OffsetDateTime}; the port's epoch-micro Instants
* round-trip exactly (PG stores microsecond TIMESTAMPTZ).
*
* @author pnoker
* @since 2026.8.20
*/
@Slf4j
public final class TimescaleTsdbStore implements TsdbStore {
private static final String TABLE = "dc3_point_value";
private static final String COLUMNS = "tenant_id, device_id, point_id, message_id, schema_version, "
+ "driver_node, sequence, fencing_token, raw_value, cal_value, num_value, quality, "
+ "driver_id, create_time, operate_time";
private final JdbcTemplate jdbc;
private static final RowMapper<PointValueSample> SAMPLE_MAPPER = (rs, i) -> new PointValueSample(
new SeriesKey(rs.getLong("tenant_id"), rs.getLong("device_id"), rs.getLong("point_id")),
toInstant(rs, "create_time"), toInstant(rs, "operate_time"),
rs.getString("raw_value"), rs.getString("cal_value"),
Objects.nonNull(rs.getObject("num_value")) ? rs.getDouble("num_value") : null,
rs.getInt("quality"),
rs.getString("message_id"), rs.getInt("schema_version"),
rs.getString("driver_node"), rs.getLong("sequence"),
rs.getLong("fencing_token"), rs.getLong("driver_id"));
public TimescaleTsdbStore(DataSource dataSource) {
this.jdbc = new JdbcTemplate(dataSource);
bootstrap();
}
private void bootstrap() {
jdbc.execute("""
CREATE TABLE IF NOT EXISTS %s (
message_id TEXT NOT NULL,
schema_version INTEGER NOT NULL,
driver_node TEXT NOT NULL,
sequence BIGINT NOT NULL,
fencing_token BIGINT NOT NULL,
device_id BIGINT NOT NULL,
point_id BIGINT NOT NULL,
raw_value TEXT NOT NULL,
cal_value TEXT NOT NULL,
num_value DOUBLE PRECISION,
quality INTEGER NOT NULL DEFAULT 0,
driver_id BIGINT NOT NULL,
tenant_id BIGINT NOT NULL,
create_time TIMESTAMPTZ NOT NULL,
operate_time TIMESTAMPTZ NOT NULL
)""".formatted(TABLE));
try {
jdbc.execute("SELECT create_hypertable('%s', 'create_time', if_not_exists => TRUE)".formatted(TABLE));
} catch (DataAccessException e) {
// timescaledb extension not loaded (plain-PG deployments): the table still
// works as a plain time-ordered table; log and continue
log.warn("TimescaleDB hypertable not created, falling back to plain table: {}", e.getMessage());
}
jdbc.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS uk_point_value_series_time
ON %s (tenant_id, device_id, point_id, create_time)""".formatted(TABLE));
jdbc.execute("""
CREATE INDEX IF NOT EXISTS idx_point_value_ts_lookup
ON %s (tenant_id, device_id, point_id, create_time DESC)""".formatted(TABLE));
jdbc.execute("""
CREATE INDEX IF NOT EXISTS idx_point_value_tenant_time
ON %s (tenant_id, create_time DESC)""".formatted(TABLE));
primeInitialChunk();
log.info("Timescale store ready (table {})", TABLE);
}
/**
* TimescaleDB sizes the initial chunk around the first inserted row; multi-row or
* microsecond-boundary first appends can leave that row outside the chunk's final
* range (invisible to index scans, present in heap). Inserting a sentinel at a
* fixed early instant forces initial chunk creation here, at bootstrap, with a
* controlled boundary — every later append lands on the normal chunk path.
*/
private void primeInitialChunk() {
try {
jdbc.update("""
INSERT INTO %s (message_id, schema_version, driver_node, sequence, fencing_token,
tenant_id, device_id, point_id, raw_value, cal_value, quality, driver_id,
create_time, operate_time)
VALUES ('dc3-chunk-prime', 1, 'bootstrap', 0, 0, 0, 0, 0, '', '', 0, 0,
'2000-01-01T00:00:00Z', '2000-01-01T00:00:00Z')
ON CONFLICT DO NOTHING""".formatted(TABLE));
jdbc.update("DELETE FROM " + TABLE + " WHERE message_id = 'dc3-chunk-prime'");
} catch (DataAccessException e) {
log.warn("Initial chunk priming skipped: {}", e.getMessage());
}
}
@Override
public String type() {
return "timescale";
}
@Override
public TsdbCapabilities capabilities() {
return new TsdbCapabilities(
true, true, true, true, true,
RollupSupport.NONE, 5000,
true, OrderingGuarantee.PER_SERIES, Precision.MICRO, true, true);
}
// ===== 写入 =====
@Override
public int append(List<PointValueSample> samples) {
if (samples.isEmpty()) {
return 0;
}
int total = 0;
for (List<PointValueSample> chunk : chunk(samples, capabilities().maxAppendBatch())) {
// TimescaleDB creates the initial chunk keyed off the FIRST row of a
// multi-row insert; when the earliest timestamp leads, that row can land
// outside the chunk's final range and become invisible to index scans.
// Sorting newest-first lets earlier timestamps backfill into the chunk the
// newest row just created — empirically verified both directions.
List<PointValueSample> ordered = new ArrayList<>(chunk);
ordered.sort(java.util.Comparator
.comparing(PointValueSample::deviceTime, java.util.Comparator.reverseOrder())
.thenComparing(PointValueSample::messageId, java.util.Comparator.reverseOrder()));
total += appendChunk(ordered);
}
return total;
}
/**
* Single-statement multi-row insert via unnest arrays — JDBC batching with
* ON CONFLICT DO UPDATE proved unreliable on the timescale-ha image (the first
* batch entry could become invisible to index scans); one statement with array
* parameters is both correct and one round trip.
*/
private int appendChunk(List<PointValueSample> chunk) {
return jdbc.execute((org.springframework.jdbc.core.ConnectionCallback<Integer>) connection -> {
String sql = """
INSERT INTO %s (%s)
SELECT * FROM unnest(
?::bigint[], ?::bigint[], ?::bigint[], ?::text[], ?::int[],
?::text[], ?::bigint[], ?::bigint[], ?::text[], ?::text[],
?::float8[], ?::int[], ?::bigint[], ?::timestamptz[], ?::timestamptz[])
ON CONFLICT (tenant_id, device_id, point_id, create_time) DO UPDATE SET
message_id = EXCLUDED.message_id,
schema_version = EXCLUDED.schema_version,
driver_node = EXCLUDED.driver_node,
sequence = EXCLUDED.sequence,
fencing_token = EXCLUDED.fencing_token,
raw_value = EXCLUDED.raw_value,
cal_value = EXCLUDED.cal_value,
num_value = EXCLUDED.num_value,
quality = EXCLUDED.quality,
driver_id = EXCLUDED.driver_id,
operate_time = EXCLUDED.operate_time""".formatted(TABLE, COLUMNS);
java.sql.Array tenant = connection.createArrayOf("bigint", longs(chunk, s -> s.series().tenantId()));
java.sql.Array device = connection.createArrayOf("bigint", longs(chunk, s -> s.series().deviceId()));
java.sql.Array point = connection.createArrayOf("bigint", longs(chunk, s -> s.series().pointId()));
java.sql.Array message = connection.createArrayOf("text", chunk.stream()
.map(PointValueSample::messageId).toArray(String[]::new));
java.sql.Array schema = connection.createArrayOf("int", chunk.stream()
.map(PointValueSample::schemaVersion).map(Integer::valueOf).toArray(Integer[]::new));
java.sql.Array node = connection.createArrayOf("text", chunk.stream()
.map(PointValueSample::driverNode).toArray(String[]::new));
java.sql.Array sequence = connection.createArrayOf("bigint", longs(chunk, PointValueSample::sequence));
java.sql.Array fencing = connection.createArrayOf("bigint", longs(chunk, PointValueSample::fencingToken));
java.sql.Array raw = connection.createArrayOf("text", chunk.stream()
.map(PointValueSample::rawValue).toArray(String[]::new));
java.sql.Array cal = connection.createArrayOf("text", chunk.stream()
.map(PointValueSample::calValue).toArray(String[]::new));
java.sql.Array num = connection.createArrayOf("float8", chunk.stream()
.map(PointValueSample::numericValue).toArray(Double[]::new));
java.sql.Array quality = connection.createArrayOf("int", chunk.stream()
.map(PointValueSample::quality).map(Integer::valueOf).toArray(Integer[]::new));
java.sql.Array driver = connection.createArrayOf("bigint", longs(chunk, PointValueSample::driverId));
java.sql.Array create = connection.createArrayOf("timestamptz", chunk.stream()
.map(s -> java.sql.Timestamp.from(s.deviceTime())).toArray(java.sql.Timestamp[]::new));
java.sql.Array operate = connection.createArrayOf("timestamptz", chunk.stream()
.map(s -> java.sql.Timestamp.from(s.receiveTime())).toArray(java.sql.Timestamp[]::new));
try (PreparedStatement ps = connection.prepareStatement(sql)) {
java.sql.Array[] arrays = {tenant, device, point, message, schema, node, sequence,
fencing, raw, cal, num, quality, driver, create, operate};
for (int i = 0; i < arrays.length; i++) {
ps.setArray(i + 1, arrays[i]);
}
return ps.executeUpdate();
}
});
}
private static Long[] longs(List<PointValueSample> chunk,
java.util.function.ToLongFunction<PointValueSample> extractor) {
return chunk.stream().mapToLong(extractor).boxed().toArray(Long[]::new);
}
// ===== 读取 =====
@Override
public Map<SeriesKey, List<PointValueSample>> last(SeriesFilter filter, int limit, TsdbDeadline deadline) {
requireSeriesOrScan(filter);
String sql = """
SELECT * FROM (
SELECT %s, ROW_NUMBER() OVER (
PARTITION BY v.tenant_id, v.device_id, v.point_id
ORDER BY v.create_time DESC, v.message_id DESC) AS rn
FROM %s v WHERE %s
) ranked WHERE rn <= ?
""".formatted(qualified(COLUMNS), TABLE, seriesWhere(filter));
List<Object> args = new ArrayList<>(seriesArgs(filter));
args.add(limit);
Map<SeriesKey, List<PointValueSample>> result = new LinkedHashMap<>();
for (PointValueSample sample : timed(deadline, () -> jdbc.query(sql, SAMPLE_MAPPER, args.toArray()))) {
result.computeIfAbsent(sample.series(), k -> new ArrayList<>()).add(sample);
}
return result;
}
@Override
public CursorPage<PointValueSample> history(SeriesFilter filter, TimeWindow window,
Cursor cursor, int pageSize, TsdbDeadline deadline) {
requireSeriesOrScan(filter);
StringBuilder sql = new StringBuilder(
"SELECT %s FROM %s v WHERE %s AND v.create_time >= ? AND v.create_time < ?"
.formatted(COLUMNS, TABLE, seriesWhere(filter)));
List<Object> args = new ArrayList<>(seriesArgs(filter));
args.add(OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
args.add(OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
if (Objects.nonNull(cursor)) {
sql.append(" AND (v.create_time, v.message_id) < (?, ?)");
args.add(OffsetDateTime.ofInstant(cursor.deviceTime(), ZoneOffset.UTC));
args.add(cursor.messageId());
}
sql.append(" ORDER BY v.create_time DESC, v.message_id DESC LIMIT ?");
args.add(pageSize + 1);
List<PointValueSample> page = timed(deadline,
() -> jdbc.query(sql.toString(), SAMPLE_MAPPER, args.toArray()));
Cursor next = null;
if (page.size() > pageSize) {
page = new ArrayList<>(page.subList(0, pageSize));
PointValueSample newest = page.get(pageSize - 1);
next = new Cursor(newest.deviceTime(), newest.messageId());
}
return new CursorPage<>(page, next);
}
@Override
public Map<SeriesKey, WindowAggregate> aggregate(SeriesFilter filter, AggregateFunction fn,
TimeWindow window, Double percentile,
TsdbDeadline deadline) {
String expr = aggregateExpression(fn, percentile);
String sql = """
SELECT tenant_id, device_id, point_id, %s AS value, COUNT(*) AS sample_count
FROM %s v WHERE %s AND create_time >= ? AND create_time < ?
GROUP BY tenant_id, device_id, point_id"""
.formatted(expr, TABLE, seriesWhere(filter));
List<Object> args = new ArrayList<>(seriesArgs(filter));
args.add(OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
args.add(OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
Map<SeriesKey, WindowAggregate> result = new LinkedHashMap<>();
timedVoid(deadline, () -> {
List<Map<String, Object>> rows = jdbc.queryForList(sql, args.toArray());
for (Map<String, Object> row : rows) {
result.put(new SeriesKey(((Number) row.get("tenant_id")).longValue(),
((Number) row.get("device_id")).longValue(),
((Number) row.get("point_id")).longValue()),
new WindowAggregate((Double) row.get("value"),
((Number) row.get("sample_count")).longValue()));
}
});
return result;
}
@Override
public Map<SeriesKey, List<BucketAggregate>> bucketedAggregate(SeriesFilter filter, AggregateFunction fn,
TimeWindow window, Duration bucketWidth,
Double percentile, TsdbDeadline deadline) {
String expr = aggregateExpression(fn, percentile);
String sql = """
SELECT tenant_id, device_id, point_id, time_bucket(?::interval, create_time) AS bucket,
%s AS value, COUNT(*) AS sample_count
FROM %s v WHERE %s AND create_time >= ? AND create_time < ?
GROUP BY tenant_id, device_id, point_id, bucket ORDER BY bucket ASC"""
.formatted(expr, TABLE, seriesWhere(filter));
List<Object> args = new ArrayList<>();
args.add(bucketWidth.toMillis() + " milliseconds");
args.addAll(seriesArgs(filter));
args.add(OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
args.add(OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
Map<SeriesKey, List<BucketAggregate>> result = new LinkedHashMap<>();
timedVoid(deadline, () -> {
List<Map<String, Object>> rows = jdbc.queryForList(sql, args.toArray());
for (Map<String, Object> row : rows) {
result.computeIfAbsent(new SeriesKey(((Number) row.get("tenant_id")).longValue(),
((Number) row.get("device_id")).longValue(),
((Number) row.get("point_id")).longValue()), k -> new ArrayList<>())
.add(new BucketAggregate(((java.sql.Timestamp) row.get("bucket")).toInstant()
.truncatedTo(java.time.temporal.ChronoUnit.MILLIS),
(Double) row.get("value"),
((Number) row.get("sample_count")).longValue()));
}
});
return result;
}
@Override
public long count(SeriesFilter filter, TimeWindow window, TsdbDeadline deadline) {
String sql = "SELECT COUNT(*) FROM " + TABLE + " v WHERE " + seriesWhere(filter)
+ " AND create_time >= ? AND create_time < ?";
List<Object> args = new ArrayList<>(seriesArgs(filter));
args.add(OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
args.add(OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
Long value = timed(deadline, () -> jdbc.queryForObject(sql, Long.class, args.toArray()));
return Objects.requireNonNullElse(value, 0L);
}
// ===== S13：租户级分析面 =====
@Override
public List<BucketAggregate> bucketedCount(long tenantId, TimeWindow window,
Duration bucketWidth, TsdbDeadline deadline) {
String sql = """
SELECT time_bucket(?::interval, create_time) AS bucket, COUNT(*) AS sample_count
FROM %s WHERE tenant_id = ? AND create_time >= ? AND create_time < ?
GROUP BY bucket ORDER BY bucket ASC""".formatted(TABLE);
Object[] args = {bucketWidth.toMillis() + " milliseconds", tenantId,
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC)};
return timed(deadline, () -> jdbc.query(sql, (rs, i) -> new BucketAggregate(
toInstant(rs, 1).truncatedTo(java.time.temporal.ChronoUnit.MILLIS),
null, rs.getLong(2)), args));
}
@Override
public List<DimensionCount> countByDimension(long tenantId, TimeWindow window,
GroupDimension dimension, int limit, TsdbDeadline deadline) {
String column = switch (dimension) {
case DEVICE -> "device_id";
case POINT -> "point_id";
case DRIVER -> "driver_id";
};
String sql = """
SELECT %s AS entity_id, COUNT(*) AS sample_count
FROM %s WHERE tenant_id = ? AND create_time >= ? AND create_time < ?
GROUP BY entity_id ORDER BY sample_count DESC LIMIT ?""".formatted(column, TABLE);
Object[] args = {tenantId,
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC),
limit};
return timed(deadline, () -> jdbc.query(sql, (rs, i) -> new DimensionCount(dimension,
rs.getLong(1), rs.getLong(2)), args));
}
@Override
public List<SeriesLastSeen> lastSeenPerSeries(long tenantId, TimeWindow window, TsdbDeadline deadline) {
String sql = """
SELECT tenant_id, device_id, point_id, MAX(create_time) AS last_seen
FROM %s WHERE tenant_id = ? AND create_time >= ? AND create_time < ?
GROUP BY tenant_id, device_id, point_id ORDER BY last_seen DESC""".formatted(TABLE);
Object[] args = {tenantId,
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC)};
return timed(deadline, () -> jdbc.query(sql, (rs, i) -> new SeriesLastSeen(
new SeriesKey(rs.getLong(1), rs.getLong(2), rs.getLong(3)), toInstant(rs, 4)), args));
}
@Override
public List<LatencyBin> latencyHistogram(long tenantId, TimeWindow window,
List<Long> binEdgesMs, TsdbDeadline deadline) {
StringBuilder bins = new StringBuilder();
for (int i = 0; i < binEdgesMs.size() + 1; i++) {
if (i < binEdgesMs.size()) {
bins.append("WHEN diff < ? THEN ").append(i).append(' ');
} else {
bins.append("ELSE ").append(i);
}
}
String sql = """
SELECT bin, COUNT(*) FROM (
SELECT CASE %s END AS bin
FROM (SELECT EXTRACT(EPOCH FROM (operate_time - create_time)) * 1000 AS diff
FROM %s
WHERE tenant_id = ? AND create_time >= ? AND create_time < ?) deltas
) bucketed GROUP BY bin ORDER BY bin"""
.formatted(bins, TABLE);
List<Object> args = new ArrayList<>();
binEdgesMs.forEach(args::add);
args.add(tenantId);
args.add(OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
args.add(OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
Map<Integer, Long> counts = new LinkedHashMap<>();
for (Map<String, Object> row : jdbc.queryForList(sql, args.toArray())) {
counts.put(((Number) row.get("bin")).intValue(), ((Number) row.get("count")).longValue());
}
List<LatencyBin> result = new ArrayList<>();
long lower = 0;
List<Long> edges = new ArrayList<>(binEdgesMs);
edges.add(Long.MAX_VALUE);
for (int i = 0; i < edges.size(); i++) {
result.add(new LatencyBin(lower, edges.get(i), counts.getOrDefault(i, 0L)));
lower = edges.get(i);
}
return result;
}
// ===== 运维 =====
@Override
public List<SeriesKey> listSeries(long tenantId, TimeWindow window, TsdbDeadline deadline) {
String sql = """
SELECT DISTINCT tenant_id, device_id, point_id FROM %s
WHERE tenant_id = ? AND create_time >= ? AND create_time < ?""".formatted(TABLE);
return timed(deadline, () -> jdbc.query(sql, (rs, i) -> new SeriesKey(
rs.getLong(1), rs.getLong(2), rs.getLong(3)),
tenantId,
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC)));
}
@Override
public void deleteRange(SeriesKey series, TimeWindow window) {
jdbc.update("DELETE FROM " + TABLE + " WHERE tenant_id = ? AND device_id = ? AND point_id = ? "
+ "AND create_time >= ? AND create_time < ?",
series.tenantId(), series.deviceId(), series.pointId(),
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
}
@Override
public CorrelationResult correlation(SeriesKey a, SeriesKey b, TimeWindow window,
Duration alignBucket, TsdbDeadline deadline) {
String sql = """
SELECT corr(a.v, b.v), count(*)
FROM (SELECT time_bucket(?::interval, create_time) AS tb, AVG(num_value) AS v FROM %s
WHERE tenant_id=? AND device_id=? AND point_id=? AND create_time>=? AND create_time<?
GROUP BY tb) a
JOIN (SELECT time_bucket(?::interval, create_time) AS tb, AVG(num_value) AS v FROM %s
WHERE tenant_id=? AND device_id=? AND point_id=? AND create_time>=? AND create_time<?
GROUP BY tb) b ON a.tb = b.tb""".formatted(TABLE, TABLE);
Object[] args = {
alignBucket.toMillis() + " milliseconds",
a.tenantId(), a.deviceId(), a.pointId(),
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC),
alignBucket.toMillis() + " milliseconds",
b.tenantId(), b.deviceId(), b.pointId(),
OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC)};
return timed(deadline, () -> jdbc.queryForObject(sql, (rs, i) -> new CorrelationResult(
rs.getDouble(1), rs.getLong(2)), args));
}
// ===== helpers =====
private static Instant toInstant(ResultSet rs, String column) throws SQLException {
Timestamp ts = rs.getTimestamp(column);
return Objects.isNull(ts) ? null : ts.toInstant();
}
private static Instant toInstant(ResultSet rs, int column) throws SQLException {
Timestamp ts = rs.getTimestamp(column);
return Objects.isNull(ts) ? null : ts.toInstant();
}
private void requireSeriesOrScan(SeriesFilter filter) {
if (filter.tenantWide() && !capabilities().tenantWideScan()) {
throw new IllegalArgumentException("tenant-wide scan not supported by this store");
}
}
private String seriesWhere(SeriesFilter filter) {
if (filter.tenantWide()) {
return "v.tenant_id = ?";
}
StringBuilder out = new StringBuilder("v.tenant_id = ? AND (");
for (int i = 0; i < filter.series().size(); i++) {
if (i > 0) {
out.append(" OR ");
}
out.append("(v.device_id = ? AND v.point_id = ?)");
}
return out.append(")").toString();
}
private List<Object> seriesArgs(SeriesFilter filter) {
List<Object> args = new ArrayList<>();
args.add(filter.tenantId());
if (!filter.tenantWide()) {
filter.series().forEach(s -> {
args.add(s.deviceId());
args.add(s.pointId());
});
}
return args;
}
private String aggregateExpression(AggregateFunction fn, Double percentile) {
return switch (fn) {
case AVG -> "AVG(num_value)";
case MIN -> "MIN(num_value)";
case MAX -> "MAX(num_value)";
case SUM -> "SUM(num_value)";
case COUNT -> "CAST(COUNT(*) AS DOUBLE PRECISION)";
case FIRST -> "(array_agg(num_value ORDER BY create_time, message_id))[1]";
case LAST -> "(array_agg(num_value ORDER BY create_time DESC, message_id DESC))[1]";
case PERCENTILE -> "percentile_cont(" + Objects.requireNonNull(percentile,
"percentile required for PERCENTILE") + ") WITHIN GROUP (ORDER BY num_value)";
};
}
private String qualified(String columns) {
StringBuilder out = new StringBuilder();
for (String column : columns.split(", ")) {
if (!out.isEmpty()) {
out.append(", ");
}
out.append("v.").append(column);
}
return out.toString();
}
private static <T> List<List<T>> chunk(List<T> list, int size) {
List<List<T>> chunks = new ArrayList<>();
for (int i = 0; i < list.size(); i += size) {
chunks.add(list.subList(i, Math.min(i + size, list.size())));
}
return chunks;
}
private void timedVoid(TsdbDeadline deadline, Runnable query) {
timed(deadline, () -> {
query.run();
return null;
});
}
private <T> T timed(TsdbDeadline deadline, java.util.function.Supplier<T> query) {
int seconds = (int) Math.max(1, deadline.maxWait().toSeconds());
Integer previous = jdbc.getQueryTimeout();
jdbc.setQueryTimeout(seconds);
try {
return query.get();
} catch (DataAccessException e) {
if (Objects.nonNull(e.getCause()) && String.valueOf(e.getCause().getClass().getName())
.contains("QueryTimeout")) {
throw new TsdbQueryTimeout("timescale query exceeded " + deadline.maxWait(), e);
}
throw e;
} finally {
jdbc.setQueryTimeout(previous);
}
}
}
dc3-tsdb/dc3-tsdb-timescale/src/main/java/io/github/pnoker/common/tsdb/timescale/config/TsdbTimescaleAutoConfiguration.java
+52
View File
@@ -0,0 +1,52 @@
/*
* Copyright 2016-present the IoT DC3 original author or authors.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package io.github.pnoker.common.tsdb.timescale.config;
import io.github.pnoker.common.tsdb.spi.TsdbStore;
import io.github.pnoker.common.tsdb.timescale.TimescaleTsdbStore;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import javax.sql.DataSource;
/**
* Activates the TimescaleDB adapter when {@code dc3.tsdb.type=timescale} (the
* default). Embedded deployments pass the primary PostgreSQL datasource; standalone
* deployments bind a dedicated one.
*
* @author pnoker
* @since 2026.8.20
*/
@Slf4j
@AutoConfiguration
@ConditionalOnClass(DataSource.class)
@ConditionalOnProperty(prefix = "dc3.tsdb", name = "type", havingValue = "timescale", matchIfMissing = true)
public class TsdbTimescaleAutoConfiguration {
@Bean
@ConditionalOnMissingBean(TsdbStore.class)
public TsdbStore tsdbStore(DataSource dataSource) {
TsdbStore store = new TimescaleTsdbStore(dataSource);
log.info("TSDB port negotiated, store={}, capabilities={}", store.type(), store.capabilities());
return store;
}
}
dc3-tsdb/dc3-tsdb-timescale/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
+18
View File
@@ -0,0 +1,18 @@
#
# Copyright 2016-present the IoT DC3 original author or authors.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
io.github.pnoker.common.tsdb.timescale.config.TsdbTimescaleAutoConfiguration
dc3-tsdb/pom.xml
+48
View File
@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2016-present the IoT DC3 original author or authors.
~
~ This program is free software: you can redistribute it and/or modify
~ it under the terms of the GNU Affero General Public License as
~ published by the Free Software Foundation, either version 3 of the
~ License, or (at your option) any later version.
~
~ This program is distributed in the hope that it will be useful,
~ but WITHOUT ANY WARRANTY; without even the implied warranty of
~ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
~ GNU Affero General Public License for more details.
~
~ You should have received a copy of the GNU Affero General Public License
~ along with this program. If not, see <https://www.gnu.org/licenses/>.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>io.github.pnoker</groupId>
<artifactId>iot-dc3</artifactId>
<version>2026.5.22</version>
</parent>
<name>${project.artifactId}</name>
<artifactId>dc3-tsdb</artifactId>
<version>2026.5.22</version>
<packaging>pom</packaging>
<description>
IoT DC3 pluggable time-series store layer: the store-neutral TsdbStore port, one
adapter per store (timescale default, tdengine, influxdb, iotdb planned) and the
contract suite that gates adapter compliance. Deployments pick exactly one store
via dc3.tsdb.type — see docs/design/tsdb-abstraction.md and docs/mq-brokers.md
for the sibling broker family.
</description>
<modules>
<module>dc3-tsdb-core</module>
<module>dc3-tsdb-timescale</module>
<module>dc3-tsdb-tck</module>
</modules>
</project>
pom.xml
+16
View File
@@ -799,6 +799,21 @@
<artifactId>dc3-mq-tck</artifactId>
<version>${dc3.version}</version>
</dependency>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb-core</artifactId>
<version>${dc3.version}</version>
</dependency>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb-timescale</artifactId>
<version>${dc3.version}</version>
</dependency>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-tsdb-tck</artifactId>
<version>${dc3.version}</version>
</dependency>
<dependency>
<groupId>io.github.pnoker</groupId>
<artifactId>dc3-common-repository</artifactId>
@@ -836,6 +851,7 @@
<module>dc3-api</module>
<module>dc3-common</module>
<module>dc3-mq</module>
<module>dc3-tsdb</module>
<module>dc3-center</module>
<module>dc3-driver</module>
<module>dc3-gateway</module>