Skip to content

Support IoTDB 2.0 REST API in table model #552

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SpriCoder merged 2 commits into from
Jul 17, 2026
Merged

Support IoTDB 2.0 REST API in table model #552

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
95848bf
Support IoTDB 2.0 REST API in table model
Jul 15, 2026
35f2847
Harden IoTDB REST handling and credential configuration
Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions configuration/conf/config.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,9 @@
# REST 接口的授权头，默认是"root:root" 的 Base64 编码结果
# REST_AUTHORIZATION=Basic cm9vdDpyb290

# REST 接口端口，默认 18080
# REST_PORT=18080

# 所有被测数据库的用户名，如果为多个数据库，则要求保持一致
# USERNAME=root

Expand Down
11 changes: 11 additions & 0 deletions core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,9 @@ public class Config {
/** Authorization header for REST interface */
private String REST_AUTHORIZATION = "Basic cm9vdDpyb290";

/** Port for REST interface */
private int REST_PORT = 18080;

// 初始化：双写模式
/** whether to operate another database */
private boolean IS_DOUBLE_WRITE = false;
Expand Down Expand Up @@ -717,6 +720,14 @@ public void setREST_AUTHORIZATION(String REST_AUTHORIZATION) {
this.REST_AUTHORIZATION = REST_AUTHORIZATION;
}

public int getREST_PORT() {
return REST_PORT;
}

public void setREST_PORT(int REST_PORT) {
this.REST_PORT = REST_PORT;
}

public long IncrementAndGetCURRENT_RECORD_LINE() {
return CURRENT_RECORD_LINE.incrementAndGet();
}
Expand Down
20 changes: 18 additions & 2 deletions core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/ConfigDescriptor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import java.util.Properties;
import java.util.stream.Collectors;

import static cn.edu.tsinghua.iot.benchmark.tsdb.enums.DBInsertMode.INSERT_USE_REST;
import static cn.edu.tsinghua.iot.benchmark.tsdb.enums.DBInsertMode.INSERT_USE_SESSION_RECORDS;
import static cn.edu.tsinghua.iot.benchmark.tsdb.enums.DBInsertMode.INSERT_USE_SESSION_TABLET;

Expand Down Expand Up @@ -105,6 +106,12 @@ private void loadProps() {
"BENCHMARK_WORK_MODE", config.getBENCHMARK_WORK_MODE() + "")));
config.setREST_AUTHORIZATION(
properties.getProperty("REST_AUTHORIZATION", config.getREST_AUTHORIZATION()));
String restAuthorization = System.getenv("BENCHMARK_REST_AUTHORIZATION");
if (restAuthorization != null && !restAuthorization.isEmpty()) {
config.setREST_AUTHORIZATION(restAuthorization);
}
config.setREST_PORT(
Integer.parseInt(properties.getProperty("REST_PORT", config.getREST_PORT() + "")));
config.setTEST_MAX_TIME(
Long.parseLong(
properties.getProperty("TEST_MAX_TIME", config.getTEST_MAX_TIME() + "")));
Expand All @@ -124,6 +131,14 @@ private void loadProps() {
config.setPORT(Arrays.asList(ports.split(",")));
config.setUSERNAME(properties.getProperty("USERNAME", config.getDbConfig().getUSERNAME()));
config.setPASSWORD(properties.getProperty("PASSWORD", config.getDbConfig().getPASSWORD()));
String benchmarkUsername = System.getenv("BENCHMARK_USERNAME");
String benchmarkPassword = System.getenv("BENCHMARK_PASSWORD");
if (benchmarkUsername != null && !benchmarkUsername.isEmpty()) {
config.setUSERNAME(benchmarkUsername);
}
if (benchmarkPassword != null && !benchmarkPassword.isEmpty()) {
config.setPASSWORD(benchmarkPassword);
}

@SpriCoder SpriCoder Jul 15, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why add this?

@FireAlbel FireAlbel Jul 15, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This component is designed to inject usernames and passwords through environment variables for cloud server-side testing, so as to enhance the robustness of configuration.

@SpriCoder SpriCoder Jul 16, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe We need to maintain this thing in docs or somewhere

@SpriCoder SpriCoder Jul 16, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

By the way, take a look at the review of copliot

@FireAlbel FireAlbel Jul 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, I have resolved the problems pointed out during the review and rolled back the modifications to the file ConfigDescriptor.java.

config.setDB_NAME(properties.getProperty("DB_NAME", config.getDbConfig().getDB_NAME()));
config.setTOKEN(properties.getProperty("TOKEN", config.getDbConfig().getTOKEN()));

Expand Down Expand Up @@ -731,9 +746,10 @@ boolean checkConfig() {
}
if (config.getIoTDB_DIALECT_MODE() == SQLDialect.TABLE
&& config.getDbConfig().getDB_SWITCH().getType() == DBType.IoTDB
&& config.getDbConfig().getDB_SWITCH().getInsertMode() != INSERT_USE_SESSION_TABLET) {
&& config.getDbConfig().getDB_SWITCH().getInsertMode() != INSERT_USE_SESSION_TABLET
&& config.getDbConfig().getDB_SWITCH().getInsertMode() != INSERT_USE_REST) {
LOGGER.error(
"The iotdb table model only supports INSERT_USE_SESSION_TABLET! Please modify DB_SWITCH in the configuration file.");
"The iotdb table model only supports INSERT_USE_SESSION_TABLET and INSERT_USE_REST! Please modify DB_SWITCH in the configuration file.");
result = false;
}
// TODO Not supported TIME_DURATION、MAX_BY、MIN_BY. iotdb will report errors for these three
Expand Down
6 changes: 6 additions & 0 deletions iotdb-2.0/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,12 @@
<artifactId>gson</artifactId>
<version>${gson.version}</version>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>mockwebserver</artifactId>
<version>${okhttp3.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<finalName>iot-benchmark-iotdb-2.0</finalName>
Expand Down
9 changes: 9 additions & 0 deletions iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDB.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,11 @@ public class IoTDB implements IDatabase {
private static final Config config = ConfigDescriptor.getInstance().getConfig();

public IoTDB(DBConfig dbConfig) throws IoTDBConnectionException {
this(dbConfig, true);
}

protected IoTDB(DBConfig dbConfig, boolean initializeDmlStrategy)
throws IoTDBConnectionException {
this.dbConfig = dbConfig;
ROOT_SERIES_NAME = "root." + dbConfig.getDB_NAME();
DELETE_SERIES_SQL = "delete storage group root." + dbConfig.getDB_NAME() + ".*";
Expand All @@ -106,6 +111,10 @@ public IoTDB(DBConfig dbConfig) throws IoTDBConnectionException {
config.getIoTDB_DIALECT_MODE() == SQLDialect.TABLE
? new TableStrategy(dbConfig)
: new TreeStrategy(dbConfig);
if (!initializeDmlStrategy) {
dmlStrategy = null;
return;
}
switch (dbConfig.getDB_SWITCH()) {
case DB_IOT_200_REST:
case DB_IOT_200_SESSION_BY_TABLET:
Expand Down
157 changes: 136 additions & 21 deletions iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPI.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import cn.edu.tsinghua.iot.benchmark.entity.Batch.IBatch;
import cn.edu.tsinghua.iot.benchmark.entity.Record;
import cn.edu.tsinghua.iot.benchmark.entity.Sensor;
import cn.edu.tsinghua.iot.benchmark.entity.enums.SQLDialect;
import cn.edu.tsinghua.iot.benchmark.exception.DBConnectException;
import cn.edu.tsinghua.iot.benchmark.measurement.Status;
import cn.edu.tsinghua.iot.benchmark.schema.schemaImpl.DeviceSchema;
Expand All @@ -25,19 +26,34 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class IoTDBRestAPI extends IoTDB {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBRestAPI.class);
private final OkHttpClient client = new OkHttpClient();
private static final Gson GSON = new Gson();
private final OkHttpClient client;
private final String baseURL;
private final DBConfig dbConfig;
protected final String ROOT_SERIES_NAME;
protected static final Config config = ConfigDescriptor.getInstance().getConfig();

public IoTDBRestAPI(DBConfig dbConfig) throws IoTDBConnectionException {
super(dbConfig);
this(dbConfig, false);
}

IoTDBRestAPI(DBConfig dbConfig, boolean skipDmlStrategy) throws IoTDBConnectionException {
super(dbConfig, !skipDmlStrategy);
this.dbConfig = dbConfig;
long timeout = config.getWRITE_OPERATION_TIMEOUT_MS();
client =
new OkHttpClient.Builder()
.connectTimeout(timeout, TimeUnit.MILLISECONDS)
.readTimeout(timeout, TimeUnit.MILLISECONDS)
.writeTimeout(timeout, TimeUnit.MILLISECONDS)
.build();
String host = dbConfig.getHOST().get(0);
baseURL = String.format("http://%s:18080", host);
ROOT_SERIES_NAME = "root";
baseURL = String.format("http://%s:%d", host, config.getREST_PORT());
ROOT_SERIES_NAME = String.format("root.%s", dbConfig.getDB_NAME());
}

private Request constructRequest(String api, String json) {
Expand All @@ -54,15 +70,16 @@ public void init() throws TsdbException {}

@Override
public void cleanup() {
String json = "{\"sql\":\"delete database root.**\"}";
Request request = constructRequest("/rest/v2/nonQuery", json);
try {
Response response = client.newCall(request).execute();
response.close();
LOGGER.info("Finish clean data!");
} catch (Exception e) {
LOGGER.warn("No Data to Clean!");
if (isTableMode()) {
for (int group = 0; group < config.getGROUP_NUMBER(); group++) {
String database =
String.format("%s_%s%d", dbConfig.getDB_NAME(), config.getGROUP_NAME_PREFIX(), group);
executeNonQuery("drop database if exists " + database);
}
} else {
executeNonQuery("delete database root." + dbConfig.getDB_NAME() + ".**");
}
LOGGER.info("Finish clean data!");
}

@Override
Expand All @@ -71,21 +88,31 @@ public void close() throws TsdbException {}
@Override
public Status insertOneBatch(IBatch batch) throws DBConnectException {
String json = generatePayload(batch);
Request request = constructRequest("/rest/v2/insertTablet", json);
String api = isTableMode() ? "/rest/table/v1/insertTablet" : "/rest/v2/insertTablet";
Request request = constructRequest(api, json);
try {
Response response = client.newCall(request).execute();
String body = response.body() == null ? "" : response.body().string();
boolean success = isSuccessful(response, body);
if (!success) {
LOGGER.warn("Insert failed: HTTP {}, body {}", response.code(), body);
}
response.close();
return new Status(true);
return new Status(success);
} catch (IOException e) {
LOGGER.warn("Insert failed!");
return new Status(false);
LOGGER.warn("Insert failed: {}", e.toString());
return new Status(false, e, "REST insert failed");
}
}

private String generatePayload(IBatch batch) {
return isTableMode() ? generateTablePayload(batch) : generateTreePayload(batch);
}

private String generateTreePayload(IBatch batch) {
DeviceSchema schema = batch.getDeviceSchema();
IoTDBRestPayload payload = new IoTDBRestPayload();
payload.device = String.format("root.%s", schema.getDevicePath());
payload.device = String.format("%s.%s", ROOT_SERIES_NAME, schema.getDevicePath());
payload.is_aligned = config.isIS_SENSOR_TS_ALIGNMENT();

List<String> measurements = new ArrayList<>();
Expand All @@ -112,7 +139,39 @@ private String generatePayload(IBatch batch) {
payload.timestamps = timestamps;
payload.values = values;

return new Gson().toJson(payload);
return GSON.toJson(payload);
}

private String generateTablePayload(IBatch batch) {
DeviceSchema schema = batch.getDeviceSchema();
IoTDBTableRestPayload payload = new IoTDBTableRestPayload();
payload.database = dbConfig.getDB_NAME() + "_" + schema.getGroup();
payload.table = schema.getTable();

for (Sensor sensor : schema.getSensors()) {
payload.column_names.add(sensor.getName());
payload.column_categories.add("FIELD");
payload.data_types.add(sensor.getSensorType().name);
}
payload.column_names.add("device_id");
payload.column_categories.add("TAG");
payload.data_types.add("STRING");
for (String key : schema.getTags().keySet()) {
payload.column_names.add(key);
payload.column_categories.add("TAG");
payload.data_types.add("STRING");
}

for (Record record : batch.getRecords()) {
payload.timestamps.add(record.getTimestamp());
List<Object> row = new ArrayList<>(record.getRecordDataValue());
row.add(schema.getDevice());
for (String key : schema.getTags().keySet()) {
row.add(schema.getTags().get(key));
}
payload.values.add(row);
}
return GSON.toJson(payload);
}

@Override
Expand All @@ -122,13 +181,16 @@ protected Status executeQueryAndGetStatus(String sql, Operation operation) {

private Status executeQueryAndGetStatus(String sql) {
String json = String.format("{\"sql\":\"%s\"}", sql);
Request request = constructRequest("/rest/v2/query", json);
String api = isTableMode() ? "/rest/table/v1/query" : "/rest/v2/query";
Request request = constructRequest(api, json);
try {
Response response = client.newCall(request).execute();
String body = response.body().string();
IoTDBRestQueryResult queryResult = new Gson().fromJson(body, IoTDBRestQueryResult.class);
IoTDBRestQueryResult queryResult = GSON.fromJson(body, IoTDBRestQueryResult.class);
response.close();
if (queryResult.timestamps == null && response.code() == 200) {
if (queryResult.timestamps == null && queryResult.values != null) {
return new Status(true, queryResult.values.size());
} else if (queryResult.timestamps == null && response.code() == 200) {
// The aggregate query has no timestamps and only one result
return new Status(true);
} else {
Expand All @@ -140,6 +202,37 @@ private Status executeQueryAndGetStatus(String sql) {
}
}

private void executeNonQuery(String sql) {
String api = isTableMode() ? "/rest/table/v1/nonQuery" : "/rest/v2/nonQuery";
String json = GSON.toJson(new IoTDBRestSql(sql));
Request request = constructRequest(api, json);
try (Response response = client.newCall(request).execute()) {
String body = response.body() == null ? "" : response.body().string();
if (!isSuccessful(response, body)) {
LOGGER.warn("Non-query failed: HTTP {}, body {}", response.code(), body);
}
} catch (IOException e) {
LOGGER.warn("Non-query failed: {}", e.getMessage());
}
}

private boolean isTableMode() {
return config.getIoTDB_DIALECT_MODE() == SQLDialect.TABLE;
}

private boolean isSuccessful(Response response, String body) {
if (!response.isSuccessful()) {
return false;
}
try {
IoTDBRestStatus status = GSON.fromJson(body, IoTDBRestStatus.class);
return status != null && status.code == 200;
} catch (RuntimeException e) {
LOGGER.warn("Invalid REST response body: {}", body);
return false;
}
}

private static class IoTDBRestPayload {
public String device;
public boolean is_aligned;
Expand All @@ -155,4 +248,26 @@ private static class IoTDBRestQueryResult {
public List<Long> timestamps;
public List<List<Object>> values;
}

private static class IoTDBRestStatus {
public int code;
}

private static class IoTDBRestSql {
public final String sql;

private IoTDBRestSql(String sql) {
this.sql = sql;
}
}

private static class IoTDBTableRestPayload {
public final List<Long> timestamps = new ArrayList<>();
public final List<String> column_names = new ArrayList<>();
public final List<String> column_categories = new ArrayList<>();
public final List<String> data_types = new ArrayList<>();
public final List<List<Object>> values = new ArrayList<>();
public String table;
public String database;
}
}
Loading
Loading