time-series: QuestDB vs InfluxDB 3 vs TimescaleDB benchmark (post 25)
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
target/
|
||||
*.class
|
||||
.idea/
|
||||
@@ -0,0 +1,20 @@
|
||||
# spring-data-demo
|
||||
|
||||
Companion code for a series of posts on [ankurm.com](https://ankurm.com) covering Spring Data integrations on Spring Boot 4.1 / JDK 25. Each top-level directory is one module, one post, independently runnable.
|
||||
|
||||
| Module | Post | Stack |
|
||||
|---|---|---|
|
||||
| [`time-series/`](./time-series) | Time-Series Data in Java: QuestDB vs InfluxDB vs TimescaleDB | QuestDB 10.0.1, InfluxDB 3 Core 3.12.0, TimescaleDB 2.25.1 (PG17) |
|
||||
| `mongodb/` | Spring Boot 4 with MongoDB: Spring Data MongoDB, Aggregations and Transactions | MongoDB 8 replica set, Spring Data MongoDB |
|
||||
| `elasticsearch/` | Full-Text Search with Spring Data Elasticsearch on Spring Boot 4 | Elasticsearch 9, Spring Data Elasticsearch |
|
||||
| `distributed-cache/` | Distributed Caching in Spring Boot: Hazelcast vs Infinispan vs Redis | Hazelcast, Infinispan, Redis |
|
||||
|
||||
Each module has its own `pom.xml`, a `docker-compose.yml` for its backing services, and a `docs/output/` directory holding the real, re-runnable transcripts that every number quoted in its post traces back to. There are no narrative doc chapters here — the deeper material for each post lives in accordions inside the post itself.
|
||||
|
||||
Build and test each module from inside its own directory, e.g.:
|
||||
|
||||
```bash
|
||||
cd time-series
|
||||
source scripts/services-up.sh # starts QuestDB, InfluxDB 3, TimescaleDB
|
||||
mvn test
|
||||
```
|
||||
@@ -0,0 +1,29 @@
|
||||
services:
|
||||
questdb:
|
||||
image: questdb/questdb:10.0.1
|
||||
ports:
|
||||
- "9009:9009" # ILP over TCP
|
||||
- "9000:9000" # HTTP / Web Console / REST
|
||||
- "8812:8812" # Postgres wire
|
||||
environment:
|
||||
- QDB_TELEMETRY_ENABLED=false
|
||||
|
||||
influxdb3:
|
||||
image: influxdb:3.12.0-core
|
||||
user: "0:0"
|
||||
ports:
|
||||
- "8181:8181"
|
||||
command: ["influxdb3", "serve", "--node-id=node0", "--object-store=file", "--data-dir=/var/lib/influxdb3/data"]
|
||||
volumes:
|
||||
- influxdb3-data:/var/lib/influxdb3/data
|
||||
|
||||
timescaledb:
|
||||
image: timescale/timescaledb:2.25.1-pg17
|
||||
ports:
|
||||
- "5433:5432"
|
||||
environment:
|
||||
- POSTGRES_PASSWORD=postgres
|
||||
- POSTGRES_DB=sensors
|
||||
|
||||
volumes:
|
||||
influxdb3-data:
|
||||
@@ -0,0 +1,3 @@
|
||||
org.apache.arrow.flight.FlightRuntimeException: Failed to read message.
|
||||
Caused by: java.lang.RuntimeException: java.lang.ClassCastException: class io.netty.buffer.PooledDirectByteBuf cannot be cast to class io.netty.buffer.PooledUnsafeDirectByteBuf (io.netty.buffer.PooledDirectByteBuf and io.netty.buffer.PooledUnsafeDirectByteBuf are in unnamed module of loader 'app')
|
||||
Caused by: java.lang.ClassCastException: class io.netty.buffer.PooledDirectByteBuf cannot be cast to class io.netty.buffer.PooledUnsafeDirectByteBuf (io.netty.buffer.PooledDirectByteBuf and io.netty.buffer.PooledUnsafeDirectByteBuf are in unnamed module of loader 'app')
|
||||
@@ -0,0 +1 @@
|
||||
{"error": "the request was not authenticated"}
|
||||
@@ -0,0 +1,4 @@
|
||||
workload: 50 devices x 2000 readings = 100000 rows
|
||||
timescaledb: ingest_ms=10048 rows=100000 rows_per_sec=9952 downsample_query_ms=49 buckets=102
|
||||
questdb: ingest_ms=302 rows=100000 rows_per_sec=331126 downsample_query_ms=41 buckets=102
|
||||
influxdb3: ingest_ms=19491 rows=100000 rows_per_sec=5131 downsample_query_ms=34 buckets=102
|
||||
@@ -0,0 +1,90 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>4.1.1</version>
|
||||
</parent>
|
||||
|
||||
<groupId>com.ankurm</groupId>
|
||||
<artifactId>time-series</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<properties>
|
||||
<java.version>25</java.version>
|
||||
<maven.compiler.release>25</maven.compiler.release>
|
||||
<questdb-client.version>1.3.8</questdb-client.version>
|
||||
<influxdb3-java.version>1.2.0</influxdb3-java.version>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-jdbc</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.postgresql</groupId>
|
||||
<artifactId>postgresql</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>tools.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- QuestDB ingestion client (ILP/QWP) - versioned independently of the QuestDB server -->
|
||||
<dependency>
|
||||
<groupId>org.questdb</groupId>
|
||||
<artifactId>questdb-client</artifactId>
|
||||
<version>${questdb-client.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- InfluxDB 3 Java client -->
|
||||
<dependency>
|
||||
<groupId>com.influxdb</groupId>
|
||||
<artifactId>influxdb3-java</artifactId>
|
||||
<version>${influxdb3-java.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers-postgresql</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers-junit-jupiter</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>testcontainers-bom</artifactId>
|
||||
<version>2.0.5</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -0,0 +1,37 @@
|
||||
#!/usr/bin/env bash
|
||||
# Brings up QuestDB 10.0.1, InfluxDB 3 Core 3.12.0 and TimescaleDB 2.25.1
|
||||
# (PG17), creates the InfluxDB 3 database + admin token, and exports
|
||||
# INFLUXDB3_TOKEN for the test run. Source this script, don't execute it:
|
||||
# source scripts/services-up.sh
|
||||
set -euo pipefail
|
||||
|
||||
cd "$(dirname "${BASH_SOURCE[0]}")/.."
|
||||
|
||||
docker compose up -d
|
||||
|
||||
echo "waiting for InfluxDB 3 Core..."
|
||||
until curl -sf http://localhost:8181/health >/dev/null 2>&1 || curl -s http://localhost:8181/health | grep -q error; do
|
||||
sleep 1
|
||||
done
|
||||
|
||||
# create (or reuse) an admin token - regenerating is fine for a local sandbox
|
||||
TOKEN_OUTPUT=$(docker exec time-series-influxdb3-1 influxdb3 create token --admin 2>&1 || true)
|
||||
export INFLUXDB3_TOKEN=$(echo "$TOKEN_OUTPUT" | grep -oP '(?<=Token: )\S+' || true)
|
||||
if [ -z "${INFLUXDB3_TOKEN:-}" ]; then
|
||||
echo "could not parse a fresh token (one may already exist) - set INFLUXDB3_TOKEN manually" >&2
|
||||
else
|
||||
docker exec time-series-influxdb3-1 influxdb3 create database sensors --token "$INFLUXDB3_TOKEN" 2>&1 || true
|
||||
echo "INFLUXDB3_TOKEN exported for this shell."
|
||||
fi
|
||||
|
||||
echo "waiting for TimescaleDB..."
|
||||
until PGPASSWORD=postgres psql -h localhost -p 5433 -U postgres -d sensors -c '\q' 2>/dev/null; do
|
||||
sleep 1
|
||||
done
|
||||
|
||||
echo "waiting for QuestDB..."
|
||||
until PGPASSWORD=quest psql -h localhost -p 8812 -U admin -d qdb -c '\q' 2>/dev/null; do
|
||||
sleep 1
|
||||
done
|
||||
|
||||
echo "all three engines are up."
|
||||
@@ -0,0 +1,121 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import com.influxdb.v3.client.InfluxDBClient;
|
||||
import com.influxdb.v3.client.Point;
|
||||
import com.influxdb.v3.client.write.WriteOptions;
|
||||
import com.influxdb.v3.client.write.WritePrecision;
|
||||
import tools.jackson.databind.JsonNode;
|
||||
import tools.jackson.databind.json.JsonMapper;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpRequest;
|
||||
import java.net.http.HttpResponse;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* InfluxDB 3 Core adapter, using the official com.influxdb:influxdb3-java
|
||||
* client for writes. There is no Spring Data module for InfluxDB 3 - this
|
||||
* client is wired in as a plain Spring bean instead.
|
||||
*
|
||||
* Queries go over the plain REST /api/v3/query_sql endpoint with a bare
|
||||
* java.net.http.HttpClient rather than the client's own queryRows()/
|
||||
* queryBatches() methods. Those methods run over Arrow Flight SQL (gRPC +
|
||||
* Netty) and throw "ClassCastException: PooledDirectByteBuf cannot be cast
|
||||
* to PooledUnsafeDirectByteBuf" against the Netty 4.2.17.Final that a
|
||||
* Spring Boot 4.1.1 + grpc-netty 1.83.1 classpath resolves - see the
|
||||
* post's "InfluxDB 3 client's query path" callout. Writes use a plain
|
||||
* HTTP line-protocol POST and are unaffected.
|
||||
*/
|
||||
public class InfluxDb3Adapter {
|
||||
|
||||
private final InfluxDBClient client;
|
||||
private final String host;
|
||||
private final String token;
|
||||
private final String database;
|
||||
private final HttpClient http = HttpClient.newHttpClient();
|
||||
private final JsonMapper json = JsonMapper.builder().build();
|
||||
|
||||
public InfluxDb3Adapter(String host, String token, String database) {
|
||||
this.client = InfluxDBClient.getInstance(host, token.toCharArray(), database);
|
||||
this.host = host;
|
||||
this.token = token;
|
||||
this.database = database;
|
||||
}
|
||||
|
||||
private List<Map<String, Object>> querySql(String sql) {
|
||||
try {
|
||||
String body = json.writeValueAsString(Map.of("db", database, "q", sql));
|
||||
HttpRequest request = HttpRequest.newBuilder()
|
||||
.uri(URI.create(host + "/api/v3/query_sql"))
|
||||
.header("Authorization", "Bearer " + token)
|
||||
.header("Content-Type", "application/json")
|
||||
.POST(HttpRequest.BodyPublishers.ofString(body))
|
||||
.build();
|
||||
HttpResponse<String> response = http.send(request, HttpResponse.BodyHandlers.ofString());
|
||||
if (response.statusCode() != 200) {
|
||||
throw new RuntimeException("query_sql failed: HTTP %d: %s".formatted(response.statusCode(), response.body()));
|
||||
}
|
||||
List<Map<String, Object>> rows = new ArrayList<>();
|
||||
for (JsonNode row : json.readTree(response.body())) {
|
||||
Map<String, Object> m = new LinkedHashMap<>();
|
||||
row.properties().forEach(e -> m.put(e.getKey(), e.getValue().isNumber() ? e.getValue().numberValue() : e.getValue().asString()));
|
||||
rows.add(m);
|
||||
}
|
||||
return rows;
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
public long ingest(List<SensorReading> readings) {
|
||||
long start = System.nanoTime();
|
||||
int batchSize = 5000;
|
||||
for (int i = 0; i < readings.size(); i += batchSize) {
|
||||
List<SensorReading> batch = readings.subList(i, Math.min(i + batchSize, readings.size()));
|
||||
List<Point> points = batch.stream().map(r ->
|
||||
Point.measurement("sensor_readings")
|
||||
.setTag("device_id", r.deviceId())
|
||||
.setTag("site", r.site())
|
||||
.setTimestamp(r.timestamp())
|
||||
.setField("temperature_c", r.temperatureC())
|
||||
.setField("humidity_pct", r.humidityPct())
|
||||
.setField("battery_volts", r.batteryVolts())
|
||||
).collect(Collectors.toList());
|
||||
client.writePoints(points, new WriteOptions.Builder()
|
||||
.precision(WritePrecision.NS)
|
||||
.build());
|
||||
}
|
||||
return (System.nanoTime() - start) / 1_000_000;
|
||||
}
|
||||
|
||||
/** Same 1-minute downsampling query, in InfluxDB 3's DataFusion-flavoured SQL. */
|
||||
public List<Map<String, Object>> downsample1m() {
|
||||
return querySql("""
|
||||
SELECT site,
|
||||
date_bin(INTERVAL '1 minute', time) AS bucket,
|
||||
avg(temperature_c) AS avg_temp,
|
||||
count(*) AS n
|
||||
FROM sensor_readings
|
||||
GROUP BY site, bucket
|
||||
ORDER BY site, bucket
|
||||
""");
|
||||
}
|
||||
|
||||
public long rowCount() {
|
||||
List<Map<String, Object>> rows = querySql("SELECT count(*) AS n FROM sensor_readings");
|
||||
return rows.isEmpty() ? 0L : ((Number) rows.get(0).get("n")).longValue();
|
||||
}
|
||||
|
||||
public void close() {
|
||||
try {
|
||||
client.close();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import io.questdb.client.Sender;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* QuestDB adapter. Ingestion uses the dedicated org.questdb:questdb-client
|
||||
* library (NOT the org.questdb:questdb artifact on Maven Central, which is
|
||||
* the embedded server engine and sits years behind the real server release
|
||||
* - see the post's "versions" callout). Queries run over QuestDB's Postgres
|
||||
* wire protocol with the same JdbcTemplate used for TimescaleDB, since
|
||||
* QuestDB speaks enough of the wire protocol for ordinary SQL + SAMPLE BY.
|
||||
*/
|
||||
public class QuestDbAdapter {
|
||||
|
||||
private final String ilpConfig;
|
||||
private final JdbcTemplate jdbc;
|
||||
|
||||
public QuestDbAdapter(String ilpConfig, JdbcTemplate jdbc) {
|
||||
this.ilpConfig = ilpConfig;
|
||||
this.jdbc = jdbc;
|
||||
}
|
||||
|
||||
public void createSchema() {
|
||||
jdbc.execute("DROP TABLE IF EXISTS sensor_readings");
|
||||
// QuestDB creates the table from the first ILP row by default (schema-on-write),
|
||||
// but we pre-create it so the column types and partitioning are explicit.
|
||||
jdbc.execute("""
|
||||
CREATE TABLE sensor_readings (
|
||||
device_id SYMBOL,
|
||||
site SYMBOL,
|
||||
temperature_c DOUBLE,
|
||||
humidity_pct DOUBLE,
|
||||
battery_volts DOUBLE,
|
||||
ts TIMESTAMP
|
||||
) TIMESTAMP(ts) PARTITION BY DAY WAL
|
||||
""");
|
||||
}
|
||||
|
||||
public long ingest(List<SensorReading> readings) {
|
||||
long start = System.nanoTime();
|
||||
try (Sender sender = Sender.fromConfig(ilpConfig)) {
|
||||
for (SensorReading r : readings) {
|
||||
sender.table("sensor_readings")
|
||||
.symbol("device_id", r.deviceId())
|
||||
.symbol("site", r.site())
|
||||
.doubleColumn("temperature_c", r.temperatureC())
|
||||
.doubleColumn("humidity_pct", r.humidityPct())
|
||||
.doubleColumn("battery_volts", r.batteryVolts())
|
||||
.at(r.timestamp());
|
||||
}
|
||||
sender.flush();
|
||||
}
|
||||
return (System.nanoTime() - start) / 1_000_000;
|
||||
}
|
||||
|
||||
/** Same 1-minute downsampling, using QuestDB's SAMPLE BY instead of time_bucket/date_bin. */
|
||||
public List<Map<String, Object>> downsample1m() {
|
||||
return jdbc.queryForList("""
|
||||
SELECT site, ts AS bucket, avg(temperature_c) AS avg_temp, count() AS n
|
||||
FROM sensor_readings
|
||||
SAMPLE BY 1m
|
||||
ALIGN TO CALENDAR
|
||||
""");
|
||||
}
|
||||
|
||||
public int rowCount() {
|
||||
Integer n = jdbc.queryForObject("SELECT count() FROM sensor_readings", Integer.class);
|
||||
return n == null ? 0 : n;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import java.time.Instant;
|
||||
|
||||
/**
|
||||
* One synthetic sensor reading used as the identical workload across
|
||||
* QuestDB, InfluxDB 3 Core and TimescaleDB. See the "Ingest throughput"
|
||||
* section of the time-series post for what this drives.
|
||||
*/
|
||||
public record SensorReading(
|
||||
String deviceId,
|
||||
String site,
|
||||
Instant timestamp,
|
||||
double temperatureC,
|
||||
double humidityPct,
|
||||
double batteryVolts
|
||||
) {
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class TimeSeriesApplication {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(TimeSeriesApplication.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
|
||||
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* TimescaleDB adapter. Plain JDBC against the Postgres wire protocol plus
|
||||
* the hypertable / time_bucket extensions Timescale adds on top of it -
|
||||
* there is no separate "Timescale client", you use spring-boot-starter-jdbc
|
||||
* (or Spring Data JDBC/JPA) exactly as you would against any Postgres.
|
||||
*/
|
||||
public class TimescaleAdapter {
|
||||
|
||||
private final JdbcTemplate jdbc;
|
||||
private final NamedParameterJdbcTemplate named;
|
||||
|
||||
public TimescaleAdapter(JdbcTemplate jdbc) {
|
||||
this.jdbc = jdbc;
|
||||
this.named = new NamedParameterJdbcTemplate(jdbc);
|
||||
}
|
||||
|
||||
public void createSchema() {
|
||||
jdbc.execute("CREATE EXTENSION IF NOT EXISTS timescaledb");
|
||||
jdbc.execute("DROP TABLE IF EXISTS sensor_readings");
|
||||
jdbc.execute("""
|
||||
CREATE TABLE sensor_readings (
|
||||
device_id TEXT NOT NULL,
|
||||
site TEXT NOT NULL,
|
||||
ts TIMESTAMPTZ NOT NULL,
|
||||
temperature_c DOUBLE PRECISION,
|
||||
humidity_pct DOUBLE PRECISION,
|
||||
battery_volts DOUBLE PRECISION
|
||||
)
|
||||
""");
|
||||
jdbc.execute("SELECT create_hypertable('sensor_readings', by_range('ts'))");
|
||||
jdbc.execute("CREATE INDEX ON sensor_readings (device_id, ts DESC)");
|
||||
}
|
||||
|
||||
public long ingest(List<SensorReading> readings) {
|
||||
long start = System.nanoTime();
|
||||
int batchSize = 2000;
|
||||
String sql = """
|
||||
INSERT INTO sensor_readings (device_id, site, ts, temperature_c, humidity_pct, battery_volts)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""";
|
||||
for (int i = 0; i < readings.size(); i += batchSize) {
|
||||
List<SensorReading> batch = readings.subList(i, Math.min(i + batchSize, readings.size()));
|
||||
jdbc.batchUpdate(sql, batch, batch.size(), (ps, r) -> {
|
||||
ps.setString(1, r.deviceId());
|
||||
ps.setString(2, r.site());
|
||||
ps.setObject(3, r.timestamp().atOffset(java.time.ZoneOffset.UTC));
|
||||
ps.setDouble(4, r.temperatureC());
|
||||
ps.setDouble(5, r.humidityPct());
|
||||
ps.setDouble(6, r.batteryVolts());
|
||||
});
|
||||
}
|
||||
return (System.nanoTime() - start) / 1_000_000;
|
||||
}
|
||||
|
||||
/** 1-minute average temperature per site, the downsampling query every engine runs. */
|
||||
public List<Map<String, Object>> downsample1m() {
|
||||
return jdbc.queryForList("""
|
||||
SELECT site,
|
||||
time_bucket('1 minute', ts) AS bucket,
|
||||
avg(temperature_c) AS avg_temp,
|
||||
count(*) AS n
|
||||
FROM sensor_readings
|
||||
GROUP BY site, bucket
|
||||
ORDER BY site, bucket
|
||||
""");
|
||||
}
|
||||
|
||||
public int rowCount() {
|
||||
Integer n = jdbc.queryForObject("SELECT count(*) FROM sensor_readings", Integer.class);
|
||||
return n == null ? 0 : n;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.SplittableRandom;
|
||||
|
||||
/** Generates the identical synthetic sensor workload for every engine under test. */
|
||||
public final class Workload {
|
||||
|
||||
private Workload() {
|
||||
}
|
||||
|
||||
public static List<SensorReading> generate(int deviceCount, int readingsPerDevice, Instant start) {
|
||||
SplittableRandom rnd = new SplittableRandom(42);
|
||||
List<SensorReading> readings = new ArrayList<>(deviceCount * readingsPerDevice);
|
||||
String[] sites = {"mumbai-dc1", "pune-dc2", "blr-dc3"};
|
||||
for (int d = 0; d < deviceCount; d++) {
|
||||
String deviceId = "sensor-%04d".formatted(d);
|
||||
String site = sites[d % sites.length];
|
||||
double baseTemp = 20 + rnd.nextDouble(10);
|
||||
for (int r = 0; r < readingsPerDevice; r++) {
|
||||
Instant ts = start.plusSeconds((long) r);
|
||||
double temp = baseTemp + Math.sin(r / 50.0) * 3 + rnd.nextDouble(-0.5, 0.5);
|
||||
double humidity = 40 + Math.cos(r / 80.0) * 10 + rnd.nextDouble(-1, 1);
|
||||
double battery = 3.7 - (r / 200_000.0);
|
||||
readings.add(new SensorReading(deviceId, site, ts, temp, humidity, battery));
|
||||
}
|
||||
}
|
||||
return readings;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import com.influxdb.v3.client.InfluxDBClient;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/**
|
||||
* Reproduces the influxdb3-java:1.2.0 query-path bug described in post 25
|
||||
* ("Time-Series Data in Java"). client.queryRows()/queryBatches() run over
|
||||
* Arrow Flight SQL (gRPC + Netty) and throw a ClassCastException against
|
||||
* the Netty 4.2.17.Final that this project's Spring Boot 4.1.1 + grpc-netty
|
||||
* 1.83.1 classpath resolves. Writes (writePoints, plain HTTP) are
|
||||
* unaffected - see InfluxDb3Adapter's REST-based querySql() workaround.
|
||||
*
|
||||
* This test is EXPECTED to throw; it exists to pin the exact exception
|
||||
* text in a committed transcript, not to assert correct behaviour.
|
||||
*/
|
||||
class ArrowFlightBugReproTest {
|
||||
|
||||
@Test
|
||||
void queryRowsThrowsOnThisClasspath() throws Exception {
|
||||
String token = System.getenv("INFLUXDB3_TOKEN");
|
||||
InfluxDBClient client = InfluxDBClient.getInstance("http://localhost:8181", token.toCharArray(), "sensors");
|
||||
|
||||
Throwable thrown = null;
|
||||
try {
|
||||
client.queryRows("SELECT 1").forEach(row -> {});
|
||||
} catch (Throwable t) {
|
||||
thrown = t;
|
||||
} finally {
|
||||
client.close();
|
||||
}
|
||||
|
||||
Transcript t = new Transcript("25-arrow-flight-bug.txt");
|
||||
if (thrown == null) {
|
||||
t.line("queryRows() did NOT throw on this classpath - bug may be fixed upstream");
|
||||
} else {
|
||||
t.line("%s: %s", thrown.getClass().getName(), thrown.getMessage());
|
||||
Throwable cause = thrown.getCause();
|
||||
while (cause != null) {
|
||||
t.line("Caused by: %s: %s", cause.getClass().getName(), cause.getMessage());
|
||||
cause = cause.getCause();
|
||||
}
|
||||
}
|
||||
t.write();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.SimpleDriverDataSource;
|
||||
import org.postgresql.Driver;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* Runs the identical ingest + 1-minute-downsample workload against QuestDB,
|
||||
* InfluxDB 3 Core and TimescaleDB and writes the real numbers to
|
||||
* docs/output/ - every figure in the time-series post traces to one of
|
||||
* these transcripts.
|
||||
*
|
||||
* Requires `docker compose up -d` to have been run first (see scripts/).
|
||||
*/
|
||||
class BenchmarkTest {
|
||||
|
||||
private static final int DEVICE_COUNT = 50;
|
||||
private static final int READINGS_PER_DEVICE = 2000; // 100,000 rows total
|
||||
private static final Instant START = Instant.parse("2026-10-14T00:00:00Z");
|
||||
|
||||
private JdbcTemplate jdbc(int port, String db, String user, String pass) {
|
||||
SimpleDriverDataSource ds = new SimpleDriverDataSource();
|
||||
ds.setDriver(new Driver());
|
||||
ds.setUrl("jdbc:postgresql://localhost:%d/%s".formatted(port, db));
|
||||
ds.setUsername(user);
|
||||
ds.setPassword(pass);
|
||||
return new JdbcTemplate(ds);
|
||||
}
|
||||
|
||||
@Test
|
||||
void ingestAndDownsampleAllThree() throws Exception {
|
||||
List<SensorReading> readings = Workload.generate(DEVICE_COUNT, READINGS_PER_DEVICE, START);
|
||||
int totalRows = readings.size();
|
||||
Transcript t = new Transcript("25-time-series-benchmark.txt");
|
||||
t.line("workload: %d devices x %d readings = %d rows", DEVICE_COUNT, READINGS_PER_DEVICE, totalRows);
|
||||
|
||||
// --- TimescaleDB ---
|
||||
JdbcTemplate tsJdbc = jdbc(5433, "sensors", "postgres", "postgres");
|
||||
TimescaleAdapter timescale = new TimescaleAdapter(tsJdbc);
|
||||
timescale.createSchema();
|
||||
long timescaleMs = timescale.ingest(readings);
|
||||
int timescaleRows = timescale.rowCount();
|
||||
long q0 = System.nanoTime();
|
||||
List<Map<String, Object>> timescaleBuckets = timescale.downsample1m();
|
||||
long timescaleQueryMs = (System.nanoTime() - q0) / 1_000_000;
|
||||
t.line("timescaledb: ingest_ms=%d rows=%d rows_per_sec=%.0f downsample_query_ms=%d buckets=%d",
|
||||
timescaleMs, timescaleRows, totalRows * 1000.0 / timescaleMs, timescaleQueryMs, timescaleBuckets.size());
|
||||
|
||||
// --- QuestDB ---
|
||||
JdbcTemplate qdbJdbc = jdbc(8812, "qdb", "admin", "quest");
|
||||
QuestDbAdapter questdb = new QuestDbAdapter("http::addr=localhost:9000;", qdbJdbc);
|
||||
questdb.createSchema();
|
||||
long questdbMs = questdb.ingest(readings);
|
||||
Thread.sleep(500); // QuestDB WAL apply is asynchronous; give it a moment before counting
|
||||
int questdbRows = questdb.rowCount();
|
||||
q0 = System.nanoTime();
|
||||
List<Map<String, Object>> questdbBuckets = questdb.downsample1m();
|
||||
long questdbQueryMs = (System.nanoTime() - q0) / 1_000_000;
|
||||
t.line("questdb: ingest_ms=%d rows=%d rows_per_sec=%.0f downsample_query_ms=%d buckets=%d",
|
||||
questdbMs, questdbRows, totalRows * 1000.0 / questdbMs, questdbQueryMs, questdbBuckets.size());
|
||||
|
||||
// --- InfluxDB 3 Core ---
|
||||
String influxToken = System.getenv("INFLUXDB3_TOKEN");
|
||||
assertTrue(influxToken != null && !influxToken.isBlank(),
|
||||
"INFLUXDB3_TOKEN env var must be set - see scripts/services-up.sh");
|
||||
InfluxDb3Adapter influx = new InfluxDb3Adapter("http://localhost:8181", influxToken, "sensors");
|
||||
long influxMs = influx.ingest(readings);
|
||||
long influxRows = influx.rowCount();
|
||||
q0 = System.nanoTime();
|
||||
List<Map<String, Object>> influxBuckets = influx.downsample1m();
|
||||
long influxQueryMs = (System.nanoTime() - q0) / 1_000_000;
|
||||
t.line("influxdb3: ingest_ms=%d rows=%d rows_per_sec=%.0f downsample_query_ms=%d buckets=%d",
|
||||
influxMs, influxRows, totalRows * 1000.0 / influxMs, influxQueryMs, influxBuckets.size());
|
||||
influx.close();
|
||||
|
||||
t.write();
|
||||
|
||||
assertEquals(totalRows, timescaleRows);
|
||||
assertEquals(totalRows, questdbRows);
|
||||
assertEquals(totalRows, influxRows);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
package com.ankurm.timeseries;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Writes docs/output/NN-*.txt while a test asserts the same numbers, so every
|
||||
* figure quoted in the blog post is backed by an assertion that fails the
|
||||
* build when it stops being true.
|
||||
*/
|
||||
final class Transcript {
|
||||
private final List<String> lines = new ArrayList<>();
|
||||
private final Path file;
|
||||
|
||||
Transcript(String name) {
|
||||
this.file = Path.of("docs/output/" + name);
|
||||
}
|
||||
|
||||
Transcript line(String s) {
|
||||
lines.add(s);
|
||||
System.out.println(s);
|
||||
return this;
|
||||
}
|
||||
|
||||
Transcript line(String fmt, Object... args) {
|
||||
return line(fmt.formatted(args));
|
||||
}
|
||||
|
||||
void write() {
|
||||
try {
|
||||
Files.createDirectories(file.getParent());
|
||||
Files.writeString(file, String.join("\n", lines) + "\n");
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user