diff --git a/configuration/conf/config.properties b/configuration/conf/config.properties index 71523f9e3..b86bcda18 100644 --- a/configuration/conf/config.properties +++ b/configuration/conf/config.properties @@ -26,6 +26,9 @@ # REST 接口的授权头，默认是"root:root" 的 Base64 编码结果 # REST_AUTHORIZATION=Basic cm9vdDpyb290 +# REST 接口端口，默认 18080 +# REST_PORT=18080 + # 所有被测数据库的用户名，如果为多个数据库，则要求保持一致 # USERNAME=root diff --git a/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/Config.java b/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/Config.java index bd52fe895..21d975482 100644 --- a/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/Config.java +++ b/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/Config.java @@ -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; @@ -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(); } diff --git a/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/ConfigDescriptor.java b/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/ConfigDescriptor.java index f069e6329..1a9955def 100644 --- a/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/ConfigDescriptor.java +++ b/core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/ConfigDescriptor.java @@ -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; @@ -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() + ""))); @@ -731,9 +738,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 diff --git a/iotdb-2.0/pom.xml b/iotdb-2.0/pom.xml index 0bcd1bb15..49eaba286 100644 --- a/iotdb-2.0/pom.xml +++ b/iotdb-2.0/pom.xml @@ -89,6 +89,12 @@ gson ${gson.version} + + com.squareup.okhttp3 + mockwebserver + ${okhttp3.version} + test + iot-benchmark-iotdb-2.0 diff --git a/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDB.java b/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDB.java index e0fd39358..51e871df6 100644 --- a/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDB.java +++ b/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDB.java @@ -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() + ".*"; @@ -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: diff --git a/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPI.java b/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPI.java index 90d229523..e4373afc0 100644 --- a/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPI.java +++ b/iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPI.java @@ -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; @@ -19,25 +20,41 @@ import okhttp3.Request; import okhttp3.RequestBody; import okhttp3.Response; +import okhttp3.ResponseBody; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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) { @@ -54,15 +71,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 @@ -71,21 +89,30 @@ public void close() throws TsdbException {} @Override public Status insertOneBatch(IBatch batch) throws DBConnectException { String json = generatePayload(batch); - Request request = constructRequest("/rest/v2/insertTablet", json); - try { - Response response = client.newCall(request).execute(); - response.close(); - return new Status(true); + String api = isTableMode() ? "/rest/table/v1/insertTablet" : "/rest/v2/insertTablet"; + Request request = constructRequest(api, json); + try (Response response = client.newCall(request).execute()) { + ResponseBody responseBody = response.body(); + String body = responseBody == null ? "" : responseBody.string(); + boolean success = isSuccessful(response, body); + if (!success) { + LOGGER.warn("Insert failed: HTTP {}, body {}", response.code(), body); + } + 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 measurements = new ArrayList<>(); @@ -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 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 @@ -122,21 +181,60 @@ 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); - try { - Response response = client.newCall(request).execute(); - String body = response.body().string(); - IoTDBRestQueryResult queryResult = new Gson().fromJson(body, IoTDBRestQueryResult.class); - response.close(); - if (queryResult.timestamps == null && response.code() == 200) { + String api = isTableMode() ? "/rest/table/v1/query" : "/rest/v2/query"; + Request request = constructRequest(api, json); + try (Response response = client.newCall(request).execute()) { + ResponseBody responseBody = response.body(); + if (responseBody == null) { + throw new IOException("REST query response body is empty"); + } + String body = responseBody.string(); + IoTDBRestQueryResult queryResult = GSON.fromJson(body, IoTDBRestQueryResult.class); + if (queryResult == null) { + throw new IOException("REST query response body is empty"); + } + 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 { return new Status(true, queryResult.timestamps.size()); } + } catch (IOException | RuntimeException e) { + LOGGER.warn("Execute query failed: {}", e.toString()); + return new Status(false, e, "REST query failed"); + } + } + + 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("Execute Query Failed!"); - return new Status(false); + 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; } } @@ -155,4 +253,26 @@ private static class IoTDBRestQueryResult { public List timestamps; public List> 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 timestamps = new ArrayList<>(); + public final List column_names = new ArrayList<>(); + public final List column_categories = new ArrayList<>(); + public final List data_types = new ArrayList<>(); + public final List> values = new ArrayList<>(); + public String table; + public String database; + } } diff --git a/iotdb-2.0/src/test/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPITest.java b/iotdb-2.0/src/test/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPITest.java new file mode 100644 index 000000000..cde2230b1 --- /dev/null +++ b/iotdb-2.0/src/test/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDBRestAPITest.java @@ -0,0 +1,232 @@ +/* + * 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 cn.edu.tsinghua.iot.benchmark.iotdb200; + +import cn.edu.tsinghua.iot.benchmark.conf.Config; +import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor; +import cn.edu.tsinghua.iot.benchmark.entity.Batch.Batch; +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.entity.enums.SensorType; +import cn.edu.tsinghua.iot.benchmark.measurement.Status; +import cn.edu.tsinghua.iot.benchmark.schema.schemaImpl.DeviceSchema; +import cn.edu.tsinghua.iot.benchmark.tsdb.DBConfig; +import cn.edu.tsinghua.iot.benchmark.tsdb.enums.DBSwitch; +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import okhttp3.OkHttpClient; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +public class IoTDBRestAPITest { + static { + System.setProperty( + "benchmark-conf", + Paths.get("..", "configuration", "conf").toAbsolutePath().normalize().toString()); + } + + private static final Config CONFIG = ConfigDescriptor.getInstance().getConfig(); + + private MockWebServer server; + private SQLDialect originalDialect; + private int originalRestPort; + private int originalWriteTimeout; + private int originalGroupNumber; + private String originalGroupPrefix; + + @Before + public void setUp() throws Exception { + originalDialect = CONFIG.getIoTDB_DIALECT_MODE(); + originalRestPort = CONFIG.getREST_PORT(); + originalWriteTimeout = CONFIG.getWRITE_OPERATION_TIMEOUT_MS(); + originalGroupNumber = CONFIG.getGROUP_NUMBER(); + originalGroupPrefix = CONFIG.getGROUP_NAME_PREFIX(); + + server = new MockWebServer(); + server.start(); + CONFIG.setIoTDB_DIALECT_MODE(SQLDialect.TABLE); + CONFIG.setREST_PORT(server.getPort()); + CONFIG.setWRITE_OPERATION_TIMEOUT_MS(4321); + CONFIG.setGROUP_NUMBER(1); + CONFIG.setGROUP_NAME_PREFIX("g_"); + } + + @After + public void tearDown() throws Exception { + CONFIG.setIoTDB_DIALECT_MODE(originalDialect); + CONFIG.setREST_PORT(originalRestPort); + CONFIG.setWRITE_OPERATION_TIMEOUT_MS(originalWriteTimeout); + CONFIG.setGROUP_NUMBER(originalGroupNumber); + CONFIG.setGROUP_NAME_PREFIX(originalGroupPrefix); + server.shutdown(); + } + + @Test + public void tableInsertUsesTableEndpointAndRowOrientedPayload() throws Exception { + server.enqueue(jsonResponse(200)); + IoTDBRestAPI restAPI = newRestAPI(); + + Status status = restAPI.insertOneBatch(newBatch()); + + assertTrue(status.isOk()); + RecordedRequest request = server.takeRequest(); + assertEquals("/rest/table/v1/insertTablet", request.getPath()); + JsonObject payload = JsonParser.parseString(request.getBody().readUtf8()).getAsJsonObject(); + assertEquals("benchmark_g_0", payload.get("database").getAsString()); + assertEquals("table_0", payload.get("table").getAsString()); + assertEquals( + Arrays.asList("s_0", "s_1", "device_id", "region"), + strings(payload.getAsJsonArray("column_names"))); + assertEquals( + Arrays.asList("FIELD", "FIELD", "TAG", "TAG"), + strings(payload.getAsJsonArray("column_categories"))); + assertEquals( + Arrays.asList("DOUBLE", "INT64", "STRING", "STRING"), + strings(payload.getAsJsonArray("data_types"))); + assertEquals(2, payload.getAsJsonArray("values").size()); + assertEquals( + "d_0", payload.getAsJsonArray("values").get(0).getAsJsonArray().get(2).getAsString()); + assertEquals( + "beijing", payload.getAsJsonArray("values").get(0).getAsJsonArray().get(3).getAsString()); + } + + @Test + public void businessErrorIsReportedAsFailureEvenWhenHttpSucceeds() throws Exception { + server.enqueue(jsonResponse(500)); + + Status status = newRestAPI().insertOneBatch(newBatch()); + + assertFalse(status.isOk()); + } + + @Test + public void tableCleanupOnlyDropsConfiguredBenchmarkDatabase() throws Exception { + server.enqueue(jsonResponse(200)); + + newRestAPI().cleanup(); + + RecordedRequest request = server.takeRequest(); + assertEquals("/rest/table/v1/nonQuery", request.getPath()); + JsonObject payload = JsonParser.parseString(request.getBody().readUtf8()).getAsJsonObject(); + assertEquals("drop database if exists benchmark_g_0", payload.get("sql").getAsString()); + } + + @Test + public void httpTimeoutUsesConfiguredWriteOperationTimeout() throws Exception { + IoTDBRestAPI restAPI = newRestAPI(); + Field clientField = IoTDBRestAPI.class.getDeclaredField("client"); + clientField.setAccessible(true); + OkHttpClient client = (OkHttpClient) clientField.get(restAPI); + + assertEquals(4321, client.connectTimeoutMillis()); + assertEquals(4321, client.readTimeoutMillis()); + assertEquals(4321, client.writeTimeoutMillis()); + } + + @Test + public void malformedQueryResponsePreservesParsingFailure() throws Exception { + server.enqueue( + new MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody("not-json")); + + Status status = executeQuery(newRestAPI(), "select * from table_0"); + + assertFalse(status.isOk()); + assertNotNull(status.getException()); + assertEquals("REST query failed", status.getErrorMessage()); + } + + @Test + public void emptyQueryResponsePreservesFailure() throws Exception { + server.enqueue(new MockResponse().setResponseCode(200).setBody("")); + + Status status = executeQuery(newRestAPI(), "select * from table_0"); + + assertFalse(status.isOk()); + assertNotNull(status.getException()); + assertEquals("REST query failed", status.getErrorMessage()); + } + + private IoTDBRestAPI newRestAPI() throws Exception { + DBConfig dbConfig = new DBConfig(); + dbConfig.setDB_SWITCH(DBSwitch.DB_IOT_200_REST); + dbConfig.setHOST(Collections.singletonList("127.0.0.1")); + dbConfig.setPORT(Collections.singletonList("6667")); + dbConfig.setDB_NAME("benchmark"); + return new IoTDBRestAPI(dbConfig, true); + } + + private Status executeQuery(IoTDBRestAPI restAPI, String sql) throws Exception { + Method method = IoTDBRestAPI.class.getDeclaredMethod("executeQueryAndGetStatus", String.class); + method.setAccessible(true); + return (Status) method.invoke(restAPI, sql); + } + + private Batch newBatch() { + Map tags = new LinkedHashMap<>(); + tags.put("region", "beijing"); + DeviceSchema schema = + new DeviceSchema( + "0", + "0", + "d_0", + Arrays.asList( + new Sensor("s_0", SensorType.DOUBLE), new Sensor("s_1", SensorType.INT64)), + tags); + return new Batch( + schema, + Arrays.asList( + new Record(1L, Arrays.asList(1.5D, 10L)), new Record(2L, Arrays.asList(2.5D, 20L)))); + } + + private MockResponse jsonResponse(int code) { + return new MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody("{\"code\":" + code + ",\"message\":\"status\"}"); + } + + private java.util.List strings(JsonArray array) { + java.util.List values = new java.util.ArrayList<>(); + array.forEach(value -> values.add(value.getAsString())); + return values; + } +}