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: + * + * + * + *

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=. + 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=). + 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; + } +}