This is an automated email from the ASF dual-hosted git repository.
CritasWang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/master by this push:
new d6245f3 Document table-level retention and refresh the README for the
merged DAOs (#117)
d6245f3 is described below
commit d6245f325304a45543eb4c64e58fc341a388c960
Author: ZIHAN DAI <[email protected]>
AuthorDate: Mon Aug 3 12:02:16 2026 +1000
Document table-level retention and refresh the README for the merged DAOs
(#117)
Add a Retention / TTL section describing how physical retention is set on
the IoTDB side: it is a table property in milliseconds, declared as
WITH (TTL=DEFAULT) in the shipped schema and changed with ALTER TABLE ...
SET PROPERTIES TTL=<ms>, readable back from information_schema.tables. It
also records the Phase-1 limitation that ThingsBoard's per-save ttl
argument cannot be a physical-retention directive, because IoTDB
retention is table-wide, so the module uses it only for storage
data-point accounting.
IoTDBTableTtlIT pins both the schema-default and the runtime-ALTER paths
against a real IoTDB 2.0.8 container. It asserts the TTL property
mechanism only, not physical row eviction, which is asynchronous and
compaction-driven and therefore not deterministic in a test.
While writing that section the surrounding README turned out to describe
the module as it was several changes ago, so this also brings it back in
line with the code: time-bucketed aggregation, latest telemetry and the
attribute DAO are implemented rather than out of scope, and key discovery
now reads DISTINCT keys from both telemetry and the telemetry_latest
overlay instead of returning an empty list. The batch findLatestByEntityIds
deferral is unchanged and stays documented.
Signed-off-by: Zihan Dai <[email protected]>
---
iotdb-thingsboard-table/README.md | 124 +++++++++---
.../extras/thingsboard/table/IoTDBTableTtlIT.java | 216 +++++++++++++++++++++
2 files changed, 316 insertions(+), 24 deletions(-)
diff --git a/iotdb-thingsboard-table/README.md
b/iotdb-thingsboard-table/README.md
index 3b08234..00f0c76 100644
--- a/iotdb-thingsboard-table/README.md
+++ b/iotdb-thingsboard-table/README.md
@@ -27,7 +27,8 @@
built on Apache IoTDB Table Mode. It lets a ThingsBoard deployment store
and serve time-series telemetry through IoTDB's table-session API instead of
the
default Cassandra/SQL backends. It compiles against the reactor's IoTDB 2.0.5
-table-session client; its integration tests run the real write path against an
+table-session client; its integration tests run the real write, read,
+aggregation, latest-telemetry, attribute and retention paths against an
`apache/iotdb:2.0.8-standalone` server. The module targets ThingsBoard
v4.3.1.2. Because
it compiles with Java 17 language features (records and others), the
`iotdb-extras` parent reactor builds and tests it only on JDK 17+ through the
@@ -58,17 +59,26 @@ DAO depends on, so any accidental edit to the local surface
fails the build.
## Scope
-This initial module delivers an inert-by-default foundation:
`IoTDBTableBaseDao`
-(session-pool lifecycle, schema/table bootstrap) and the
-`IoTDBTableTimeseriesDao` write path (`save`), raw read, and delete. To
exercise
-it, set both `database.ts.type=iotdb-table` and
-`iotdb.ts.experimental-raw-only=true`. Aggregation, latest telemetry, and
-attribute/label DAOs are outside the current scope.
-
-> **This is an incremental / experimental backend.** Explicitly enabling it
-> routes ThingsBoard historical telemetry through IoTDB Table Mode for **raw
read
-> + write + delete only**. **Time-bucketed aggregation is NOT implemented
yet**;
-> aggregation, latest telemetry, and attributes are outside the current scope.
+The module is an inert-by-default foundation — `IoTDBTableBaseDao`
(session-pool
+lifecycle, schema/table bootstrap) — plus three DAOs, each behind its own
+selector:
+
+- `IoTDBTableTimeseriesDao`: the historical write path (`save`), raw read,
+ delete, and time-bucketed aggregation — both the fixed-width millisecond
+ `date_bin` path and the timezone-aware calendar path (`WEEK` / `WEEK_ISO` /
+ `MONTH` / `QUARTER`). Enabled by `database.ts.type=iotdb-table` together with
+ `iotdb.ts.experimental-raw-only=true`. That property name predates the
+ aggregation support and is kept for compatibility; it is the opt-in for this
+ backend as a whole, not a raw-only switch.
+- `IoTDBTableLatestDao`: latest telemetry, derived from `telemetry` with a
+ `telemetry_latest` overlay. Enabled by `database.ts_latest.type=iotdb-table`
+ (see the latest-telemetry section below).
+- `IoTDBTableAttributesDao`: entity attributes, **inert by default**. Enabled
+ only by the independent `database.attributes.type=iotdb-table` opt-in.
+
+> **This is an incremental / experimental backend.** Nothing routes through
+> IoTDB Table Mode unless the matching selector is set explicitly; with no
+> selectors set the module does nothing.
## Entity attributes (inert by default)
@@ -188,19 +198,82 @@ is the overlay's own already-stored value is reported
without a redundant rewrit
identity; bounded for normal key sets).
- **Same-timestamp cross-store type change.** The overlay wins an exact-ts tie,
continuing the documented same-timestamp (B1) limitation below.
-- **Batch latest read / key discovery deferred (graceful, not throwing).**
- `findLatestByEntityIds(Async)` (new in v4.3.1.2, full impl is a follow-up)
and the
- key-discovery methods `findAllKeysByEntityIds(Async)` (a follow-up) return
an **empty
- list** rather than throwing, because they are reachable in normal operation
(the
- dashboard `/api/entitiesQuery/find/keys` lookup and entity-delete
housekeeping),
- where a throw would surface as an HTTP 500 / a failed cleanup task. This
matches
- the official `CassandraBaseTimeseriesLatestDao`, which returns empty for all
four.
+- **Batch latest read deferred (graceful, not throwing).**
+ `findLatestByEntityIds(Async)` (new in v4.3.1.2) returns an **empty list**
+ rather than throwing, because it is reachable in normal operation and a throw
+ would surface as an HTTP 500. This matches the official
+ `CassandraBaseTimeseriesLatestDao`; the derived batch read is a follow-up.
+ Key discovery (`findAllKeysByEntityIds(Async)` and the tenant-wide variant)
is
+ implemented: it collects `DISTINCT key` from **both** `telemetry` and the
+ `telemetry_latest` overlay, so a key that only ever reached the overlay
through
+ `saveLatest` is discovered too.
The `telemetry_latest` table is created on startup by a **second** idempotent
schema bootstrap (`schema-iotdb-table-latest.sql`), registered only when the
latest selector is active and `iotdb.schema.bootstrap` is not disabled; when
the
latest selector is off the overlay table is never created.
+## Retention / TTL
+
+Physical retention is a **table-level** IoTDB property that the operator sets
on
+the schema; it is not a per-data-point setting. IoTDB Table Mode expresses TTL
as
+a retention window in **milliseconds**, and the accepted forms are narrow:
+
+| Form | Meaning |
+| --- | --- |
+| `TTL=604800000` | A concrete retention: a bare, **unquoted** long literal
(milliseconds). |
+| `TTL='INF'` | Never expire. The **quoted** string is the only accepted
spelling; this is the form `entity_attributes` and `telemetry_latest` ship
with. |
+| `TTL=DEFAULT` | Inherit the database default, which is `INF` on a fresh
node. |
+
+Anything else is rejected by IoTDB 2.0.8: an unquoted `TTL=INF` is parsed as an
+identifier (`ttl value must be a LongLiteral, but now is Identifier`), and any
+other quoted value — including a quoted number (`'604800000'`) or a duration
+(`'7d'`) — fails with `ttl value must be 'INF' or a long literal`.
+
+The shipped `schema-iotdb-table.sql` declares `telemetry` with
+`WITH (TTL=DEFAULT)`, so a fresh deployment never expires data until an
operator
+chooses otherwise. To set a concrete retention, either edit the schema before
the
+first bootstrap:
+
+```sql
+CREATE TABLE telemetry (...) WITH (TTL=604800000);
+```
+
+or change it at runtime on the live table:
+
+```sql
+ALTER TABLE telemetry SET PROPERTIES TTL=604800000; -- 7 days, in ms
+ALTER TABLE telemetry SET PROPERTIES TTL=DEFAULT; -- back to the db default
+```
+
+The effective value can be read back from `SHOW TABLES`, which reports it in a
+`TTL(ms)` column, or from `information_schema.tables`, where the column must be
+quoted because its name contains parentheses:
+
+```sql
+SELECT table_name, "ttl(ms)" FROM information_schema.tables WHERE
database='thingsboard';
+```
+
+Either way a never-expiring table reads back as `INF` and a concrete retention
as
+the millisecond number.
+`IoTDBTableTtlIT` pins both paths against a real IoTDB 2.0.8 container. It
+verifies the TTL **property mechanism** only and deliberately does not assert
+physical row eviction, because eviction is asynchronous and compaction-driven
and
+so is not deterministic inside a test.
+
+The `telemetry_latest` overlay is declared `TTL='INF'` on purpose: it holds at
+most one row per identity and is the authority for a latest value, so expiring
it
+would silently drop the latest reading while history remained.
+
+- **Phase-1 limitation: the per-save `ttl` argument is not a physical-retention
+ directive.** ThingsBoard's `TimeseriesDao.save(..., long ttl)` carries a
+ per-data-point TTL, but IoTDB Table Mode retention can only be expressed at
the
+ table level, so the two cannot be reconciled faithfully. The module therefore
+ uses that argument (together with `iotdb.defaultTtlMs`) **only** for
+ ThingsBoard's storage data-point accounting, and never as an instruction to
+ expire rows. Operators who need physical retention set it on the table as
shown
+ above.
+
## Known limitations
**Same-timestamp type change across separate flushes.** The writer collapses
@@ -228,7 +301,7 @@ Key activation and operational flags:
| Property | Default | Meaning |
| --- | --- | --- |
| `database.ts.type` | _(unset)_ | Set to `iotdb-table` as the ThingsBoard
historical-timeseries backend selector. |
-| `iotdb.ts.experimental-raw-only` | `false` | Explicit opt-in for this
initial raw-only backend. Must be `true` together with
`database.ts.type=iotdb-table`; write, raw read, and delete are implemented,
while time-bucketed aggregation is outside the current scope. |
+| `iotdb.ts.experimental-raw-only` | `false` | Explicit opt-in for this
backend. Must be `true` together with `database.ts.type=iotdb-table`. The name
predates the aggregation support and is kept for compatibility: write, raw
read, delete **and** time-bucketed aggregation are all served when it is
enabled. |
| `database.attributes.type` | _(unset)_ | Set to `iotdb-table` to opt in to
the entity-attribute DAO. Independent of the timeseries selectors. Unset in a
real Phase-1 deployment, so the attribute DAO is inert by default. |
| `iotdb.attributes.cluster_mode` | _(empty)_ | Required when
`database.attributes.type=iotdb-table`. Must be `sticky-routing` (per-identity
writes pinned to one node) or `disabled` (single-node / acknowledged
best-effort); any other value (including the empty default) fails construction
fast, because the attribute write path converges only within a single JVM. |
| `iotdb.ts_latest.cluster_mode` | _(empty)_ | Required when
`database.ts_latest.type=iotdb-table` (the latest-overlay DAO is active). Must
be `sticky-routing` (per-identity latest writes pinned to one node) or
`disabled` (single-node / acknowledged best-effort); any other value (including
the empty default) fails construction fast, because the latest-overlay write
path converges only within a single JVM. This is the symmetric acknowledgement
to `iotdb.attributes.cluster_mode`. |
@@ -298,11 +371,14 @@ docker compose -f docker-compose.test.yml down -v
## Status
-Initial module status: `IoTDBTableBaseDao` plus the `IoTDBTableTimeseriesDao`
-write, raw-read, and delete paths are implemented behind
+`IoTDBTableBaseDao` plus the `IoTDBTableTimeseriesDao` write, raw-read, delete
+and time-bucketed aggregation paths are implemented behind
`database.ts.type=iotdb-table` and `iotdb.ts.experimental-raw-only=true`.
-Without both properties, the module is inert. Aggregation and latest telemetry
-are outside the current scope.
+Without both properties, that DAO is inert. `IoTDBTableLatestDao` (latest
+telemetry, with the `telemetry_latest` overlay and key discovery) is
implemented
+behind its own `database.ts_latest.type=iotdb-table` selector. Physical
+retention is a table property the operator sets on the schema; see
+Retention / TTL above.
`IoTDBTableAttributesDao` is **inert by default** and activated only by the
independent
`database.attributes.type=iotdb-table` opt-in (see the Entity attributes
section
diff --git
a/iotdb-thingsboard-table/src/test/java/org/apache/iotdb/extras/thingsboard/table/IoTDBTableTtlIT.java
b/iotdb-thingsboard-table/src/test/java/org/apache/iotdb/extras/thingsboard/table/IoTDBTableTtlIT.java
new file mode 100644
index 0000000..bd6cf4f
--- /dev/null
+++
b/iotdb-thingsboard-table/src/test/java/org/apache/iotdb/extras/thingsboard/table/IoTDBTableTtlIT.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iotdb.extras.thingsboard.table;
+
+import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.pool.ITableSessionPool;
+import org.apache.iotdb.session.pool.TableSessionPoolBuilder;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Validates the table-level TTL mechanism against real IoTDB 2.0.8. IoTDB
Table Mode TTL is a table
+ * property expressed in milliseconds; this IT proves the two operator-facing
paths the README
+ * documents work as described:
+ *
+ * <ul>
+ * <li>the schema's {@code WITH (TTL=DEFAULT)} bootstraps and resolves to
the database default
+ * ({@code INF}, i.e. never expire), and
+ * <li>{@code ALTER TABLE telemetry SET PROPERTIES TTL=<ms>} sets a concrete
millisecond retention
+ * at runtime, and {@code CREATE TABLE ... WITH (TTL=<ms>)} sets it at
create time.
+ * </ul>
+ *
+ * <p>The asserted TTL is read back from {@code information_schema.tables}
(the {@code ttl(ms)}
+ * column). This validates the TTL DDL/property mechanism only. It
deliberately does NOT assert
+ * physical row eviction: IoTDB TTL eviction is async and compaction-driven,
so observed row drops
+ * are not deterministic within a test and are out of scope here.
+ */
+@Tag("integration")
+@Testcontainers(disabledWithoutDocker = true)
+class IoTDBTableTtlIT {
+ private static final Duration IOTDB_STARTUP_TIMEOUT = Duration.ofMinutes(3);
+ private static final Duration IOTDB_READY_TIMEOUT = Duration.ofSeconds(60);
+ private static final Duration IOTDB_READY_POLL_INTERVAL =
Duration.ofMillis(500);
+
+ // 7 days expressed in milliseconds: IoTDB Table Mode TTL is a bare long
literal in ms.
+ private static final long SEVEN_DAYS_MS = TimeUnit.DAYS.toMillis(7);
+ // 1 day in ms, used for the CREATE-time path.
+ private static final long ONE_DAY_MS = TimeUnit.DAYS.toMillis(1);
+
+ @Container
+ static final GenericContainer<?> IOTDB =
+ new
GenericContainer<>(DockerImageName.parse("apache/iotdb:2.0.8-standalone"))
+ .withExposedPorts(6667)
+ // IoTDB binds its client RPC service to dn_rpc_address (default
127.0.0.1); bind to all
+ // interfaces so the Testcontainers port-mapped session handshake
succeeds.
+ .withEnv("dn_rpc_address", "0.0.0.0")
+
.waitingFor(Wait.forListeningPort().withStartupTimeout(IOTDB_STARTUP_TIMEOUT));
+
+ @Test
+ void
schemaBootstrapsWithDefaultTtlThenAlterSetsConcreteMillisecondRetention()
throws Exception {
+ String database = uniqueDatabase("ttl_alter");
+ bootstrapSchema(database);
+ try (ITableSessionPool pool = newPool(database)) {
+ // The shipped schema declares telemetry WITH (TTL=DEFAULT); DEFAULT
resolves to the database
+ // default retention, which on a fresh node is INF (never expire).
+ assertEquals("INF", readTableTtlMs(pool, database, "telemetry"));
+
+ // The same schema declares entity_attributes WITH (TTL='INF'). The
QUOTED string is the only
+ // accepted never-expire literal: an unquoted INF is parsed as an
identifier and rejected with
+ // "ttl value must be a LongLiteral, but now is Identifier". Asserting
the read-back pins the
+ // exact form the shipped schema depends on, not just that the bootstrap
did not throw.
+ assertEquals("INF", readTableTtlMs(pool, database, "entity_attributes"));
+
+ // Operator switches to a concrete retention at runtime via ALTER ...
SET PROPERTIES TTL=<ms>.
+ try (ITableSession session = pool.getSession()) {
+ session.executeNonQueryStatement(
+ "ALTER TABLE telemetry SET PROPERTIES TTL=" + SEVEN_DAYS_MS);
+ }
+ assertEquals(Long.toString(SEVEN_DAYS_MS), readTableTtlMs(pool,
database, "telemetry"));
+
+ // And can revert to the database default (INF here) the same way.
+ try (ITableSession session = pool.getSession()) {
+ session.executeNonQueryStatement("ALTER TABLE telemetry SET PROPERTIES
TTL=DEFAULT");
+ }
+ assertEquals("INF", readTableTtlMs(pool, database, "telemetry"));
+ }
+ }
+
+ @Test
+ void createTableWithConcreteMillisecondTtlIsReadBackExactly() throws
Exception {
+ String database = uniqueDatabase("ttl_create");
+ bootstrapSchema(database);
+ try (ITableSessionPool pool = newPool(database)) {
+ // The schema can equally ship a concrete retention at create time: WITH
(TTL=<ms>).
+ try (ITableSession session = pool.getSession()) {
+ session.executeNonQueryStatement(
+ "CREATE TABLE telemetry_ttl ("
+ + "tenant_id STRING TAG, entity_type STRING TAG, entity_id
STRING TAG, "
+ + "key STRING TAG, long_v INT64 FIELD) WITH (TTL="
+ + ONE_DAY_MS
+ + ")");
+ }
+ assertEquals(Long.toString(ONE_DAY_MS), readTableTtlMs(pool, database,
"telemetry_ttl"));
+ }
+ }
+
+ /**
+ * Reads the table's TTL back from {@code information_schema.tables}. The
{@code ttl(ms)} column
+ * is the millisecond retention, or the literal {@code INF} when the table
never expires.
+ */
+ private String readTableTtlMs(ITableSessionPool pool, String database,
String table)
+ throws Exception {
+ String sql =
+ "SELECT \"ttl(ms)\" FROM information_schema.tables "
+ + "WHERE database='"
+ + database
+ + "' AND table_name='"
+ + table
+ + "'";
+ try (ITableSession session = pool.getSession();
+ SessionDataSet dataSet = session.executeQueryStatement(sql)) {
+ SessionDataSet.DataIterator row = dataSet.iterator();
+ assertTrue(row.next(), "expected one information_schema row for table "
+ table);
+ return row.getString("ttl(ms)");
+ }
+ }
+
+ private ITableSessionPool newPool(String database) {
+ TableSessionPoolBuilder builder =
+ new TableSessionPoolBuilder()
+ .nodeUrls(List.of("127.0.0.1:" + IOTDB.getMappedPort(6667)))
+ .user("root")
+ .password("root")
+ .maxSize(2);
+ if (database != null) {
+ builder.database(database);
+ }
+ return builder.build();
+ }
+
+ private void bootstrapSchema(String database) throws Exception {
+ awaitIoTDBReady(database);
+
+ String schema;
+ try (InputStream stream =
+
IoTDBTableTtlIT.class.getClassLoader().getResourceAsStream("schema-iotdb-table.sql"))
{
+ schema = new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ schema =
+ schema
+ .replace(
+ "CREATE DATABASE IF NOT EXISTS thingsboard;",
+ "CREATE DATABASE IF NOT EXISTS " + database + ";")
+ .replace("USE thingsboard;", "USE " + database + ";");
+ schema = schema.replaceAll("(?s)/\\*.*?\\*/", "").replaceAll("(?m)--.*$",
"");
+ try (ITableSessionPool bootstrapPool = newPool(null);
+ ITableSession session = bootstrapPool.getSession()) {
+ for (String statement : schema.split(";")) {
+ String trimmed = statement.trim();
+ if (!trimmed.isEmpty()) {
+ session.executeNonQueryStatement(trimmed);
+ }
+ }
+ }
+ }
+
+ private void awaitIoTDBReady(String database) throws Exception {
+ long deadlineNanos = System.nanoTime() + IOTDB_READY_TIMEOUT.toNanos();
+ Exception lastFailure = null;
+ while (System.nanoTime() < deadlineNanos) {
+ try (ITableSessionPool bootstrapPool = newPool(null);
+ ITableSession session = bootstrapPool.getSession()) {
+ session.executeNonQueryStatement("CREATE DATABASE IF NOT EXISTS " +
database);
+ return;
+ } catch (Exception e) {
+ lastFailure = e;
+ long remainingMillis = TimeUnit.NANOSECONDS.toMillis(deadlineNanos -
System.nanoTime());
+ if (remainingMillis <= 0) {
+ break;
+ }
+ Thread.sleep(Math.min(IOTDB_READY_POLL_INTERVAL.toMillis(),
remainingMillis));
+ }
+ }
+ throw new IllegalStateException(
+ "IoTDB did not accept table-session statements within " +
IOTDB_READY_TIMEOUT, lastFailure);
+ }
+
+ private String uniqueDatabase(String prefix) {
+ String shortPrefix = prefix.length() > 12 ? prefix.substring(0, 12) :
prefix;
+ String shortUuid = UUID.randomUUID().toString().replace("-",
"").substring(0, 16);
+ return "tb_it_" + shortPrefix + "_" + shortUuid;
+ }
+}
