Add a base mode that can execute a specified SQL statement and benchmark/measure the query latency metrics

configuration/conf/config.properties

@@ -55,6 +55,7 @@
 # generateDataMode      生成数据模式，生成Benchmark本身识别的数据
 # verificationWriteMode 单数据库正确性写入模式，需要配置 FILE_PATH 以及 DATA_SET
 # verificationQueryMode 单数据库正确性查询模式，需要配置 FILE_PATH 以及 DATA_SET
+# fixedQueryMode 固定查询指定SQL模式，需要配置 FIXED_QUERY_SQL 以及 QUERY_LOOP 和 OP_MIN_INTERVAL
 # BENCHMARK_WORK_MODE=testWithDefaultPath
 
 # 对于数据写入或查询，限制最长耗时，设置为0表示无限制，单位为ms
@@ -502,3 +503,9 @@
 # 测试过程中当前测试结果日志的输出间隔，单位为秒
 # 设为0时，benchmark 仅在结束前输出一次测试结果日志
 # RESULT_PRINT_INTERVAL=3600
+
+############# 固定查询指定SQL模式参数 #######################
+# 指定执行的SQL
+# FIXED_QUERY_SQL=SELECT data_node_id, client_ip FROM information_schema.connections
+# 每个客户端执行的循环次数
+# QUERY_LOOP=10

core/src/main/java/cn/edu/tsinghua/iot/benchmark/App.java

@@ -23,6 +23,7 @@
 import cn.edu.tsinghua.iot.benchmark.conf.ConfigDescriptor;
 import cn.edu.tsinghua.iot.benchmark.measurement.persistence.csv.CSVShutdownHook;
 import cn.edu.tsinghua.iot.benchmark.mode.BaseMode;
+import cn.edu.tsinghua.iot.benchmark.mode.FixedQueryMode;
 import cn.edu.tsinghua.iot.benchmark.mode.GenerateDataMode;
 import cn.edu.tsinghua.iot.benchmark.mode.TestWithDefaultPathMode;
 import cn.edu.tsinghua.iot.benchmark.mode.VerificationQueryMode;
@@ -64,6 +65,9 @@ public static void main(String[] args) throws SQLException {
       case VERIFICATION_QUERY:
         baseMode = new VerificationQueryMode();
         break;
+      case FIXED_QUERY:
+        baseMode = new FixedQueryMode();
+        break;
       default:
         throw new SQLException("Unsupported mode:" + config.getBENCHMARK_WORK_MODE());
     }

core/src/main/java/cn/edu/tsinghua/iot/benchmark/client/DataClient.java

@@ -23,6 +23,7 @@
 import cn.edu.tsinghua.iot.benchmark.client.generate.GenerateDataMixClient;
 import cn.edu.tsinghua.iot.benchmark.client.generate.GenerateDataWriteClient;
 import cn.edu.tsinghua.iot.benchmark.client.progress.TaskProgress;
+import cn.edu.tsinghua.iot.benchmark.client.real.FixedQueryClient;
 import cn.edu.tsinghua.iot.benchmark.client.real.RealDataSetQueryClient;
 import cn.edu.tsinghua.iot.benchmark.client.real.RealDataSetWriteClient;
 import cn.edu.tsinghua.iot.benchmark.conf.Config;
@@ -104,6 +105,8 @@ public static DataClient getInstance(
         return new RealDataSetWriteClient(id, countDownLatch, barrier, taskProgress);
       case VERIFICATION_QUERY:
         return new RealDataSetQueryClient(id, countDownLatch, barrier, taskProgress);
+      case FIXED_QUERY:
+        return new FixedQueryClient(id, countDownLatch, barrier, taskProgress);
       default:
         LOGGER.warn("No need to create client" + config.getBENCHMARK_WORK_MODE());
         break;

core/src/main/java/cn/edu/tsinghua/iot/benchmark/client/real/FixedQueryClient.java

@@ -0,0 +1,78 @@
+/*
+ * 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.client.real;
+
+import cn.edu.tsinghua.iot.benchmark.client.progress.TaskProgress;
+import cn.edu.tsinghua.iot.benchmark.measurement.Status;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+
+public class FixedQueryClient extends RealBaseClient {
+
+  private static final int queryLoop = config.getQUERY_LOOP();
+
+  public FixedQueryClient(
+      int id, CountDownLatch countDownLatch, CyclicBarrier barrier, TaskProgress taskProgress) {
+    super(id, countDownLatch, barrier, taskProgress);
+  }
+
+  /** Do Operations */
+  @Override
+  protected void doTest() {
+    for (int i = 0; i < queryLoop; i++) {
+      try {
+        Status status = dbWrapper.deviceQuery(config.getFIXED_QUERY_SQL());
+        long start = 0;
+        if (config.getOP_MIN_INTERVAL() > 0) {
+          start = System.currentTimeMillis();
+        }
+
+        if (!status.isOk()) {
+          return;
+        }
+
+        double costTime = status.getTimeCost() / Math.pow(10, 6);
+        LOGGER.debug(
+            "sql cost time {} ms, sql is {}, result is {}",
+            costTime,
+            config.getFIXED_QUERY_SQL(),
+            status.getRecords());
+        taskProgress.incrementLoopIndex();
+
+        if (config.getOP_MIN_INTERVAL() > 0) {
+          long opMinInterval;
+          opMinInterval = config.getOP_MIN_INTERVAL();
+          long elapsed = System.currentTimeMillis() - start;
+          if (elapsed < opMinInterval) {
+            try {
+              LOGGER.debug("[Client-{}] sleep {} ms.", clientThreadId, opMinInterval - elapsed);
+              Thread.sleep(opMinInterval - elapsed);
+            } catch (InterruptedException e) {
+              LOGGER.error("Wait for next operation failed because ", e);
+            }
+          }
+        }
+      } catch (Exception e) {
+        LOGGER.error("Failed to query sql because ", e);
+      }
+    }
+  }
+}

core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/Config.java

@@ -535,6 +535,11 @@ public class Config {
   /** Sensor function */
   private Map<String, FunctionParam> SENSOR_FUNCTION = new HashMap<>();
 
+  /** fixed query mode config parameters */
+  private String FIXED_QUERY_SQL = "";
+
+  private int QUERY_LOOP = 10;
+
   /** init inner functions */
   public void initInnerFunction() {
     FunctionXml xml = null;
@@ -1865,6 +1870,22 @@ public void setIS_RECORD_CURRENT_REALLY_TIME(boolean IS_RECORD_CURRENT_REALLY_TI
     this.IS_RECORD_CURRENT_REALLY_TIME = IS_RECORD_CURRENT_REALLY_TIME;
   }
 
+  public String getFIXED_QUERY_SQL() {
+    return FIXED_QUERY_SQL;
+  }
+
+  public void setFIXED_QUERY_SQL(String FIXED_QUERY_SQL) {
+    this.FIXED_QUERY_SQL = FIXED_QUERY_SQL;
+  }
+
+  public int getQUERY_LOOP() {
+    return QUERY_LOOP;
+  }
+
+  public void setQUERY_LOOP(int QUERY_LOOP) {
+    this.QUERY_LOOP = QUERY_LOOP;
+  }
+
   /** write dataset config to info */
   public String toInfoText() {
     return "LOOP="

core/src/main/java/cn/edu/tsinghua/iot/benchmark/conf/ConfigDescriptor.java

@@ -568,6 +568,12 @@ private void loadProps() {
                 properties.getProperty(
                     "IS_RECORD_CURRENT_REALLY_TIME",
                     config.isIS_RECORD_CURRENT_REALLY_TIME() + "")));
+
+        config.setFIXED_QUERY_SQL(
+            properties.getProperty("FIXED_QUERY_SQL", config.getFIXED_QUERY_SQL()));
+
+        config.setQUERY_LOOP(
+            Integer.parseInt(properties.getProperty("QUERY_LOOP", config.getQUERY_LOOP() + "")));
       } catch (IOException e) {
         e.printStackTrace();
       }

core/src/main/java/cn/edu/tsinghua/iot/benchmark/mode/FixedQueryMode.java

@@ -0,0 +1,42 @@
+/*
+ * 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.mode;
+
+import cn.edu.tsinghua.iot.benchmark.client.DataClient;
+import cn.edu.tsinghua.iot.benchmark.client.operation.Operation;
+
+import java.util.Collections;
+
+public class FixedQueryMode extends BaseMode {
+
+  @Override
+  protected boolean preCheck() {
+    return true;
+  }
+
+  @Override
+  protected void postCheck() {
+    finalMeasure(
+        baseModeMeasurement,
+        dataClients.stream().map(DataClient::getMeasurement),
+        startTime,
+        Collections.singletonList(Operation.DEVICE_QUERY));
+  }
+}

core/src/main/java/cn/edu/tsinghua/iot/benchmark/mode/enums/BenchmarkMode.java

@@ -26,7 +26,8 @@ public enum BenchmarkMode {
   TEST_WITH_DEFAULT_PATH("testWithDefaultPath"),
   GENERATE_DATA("generateDataMode"),
   VERIFICATION_WRITE("verificationWriteMode"),
-  VERIFICATION_QUERY("verificationQueryMode");
+  VERIFICATION_QUERY("verificationQueryMode"),
+  FIXED_QUERY("fixedQueryMode");
 
   private static final Logger LOGGER = LoggerFactory.getLogger(BenchmarkMode.class);
   public String mode;

core/src/main/java/cn/edu/tsinghua/iot/benchmark/tsdb/DBWrapper.java

@@ -469,6 +469,28 @@ public Status deviceQuery(DeviceQuery deviceQuery) throws SQLException {
     return status;
   }
 
+  @Override
+  public Status deviceQuery(String sql) {
+    Status status = null;
+    Operation operation = Operation.DEVICE_QUERY;
+    try {
+      List<Status> statuses = new ArrayList<>();
+      for (IDatabase database : databases) {
+        long start = System.nanoTime();
+        status = database.deviceQuery(sql);
+        long end = System.nanoTime();
+        status.setTimeCost(end - start);
+        statuses.add(status);
+      }
+      for (Status sta : statuses) {
+        handleQueryOperation(sta, operation, "");
+      }
+    } catch (Exception e) {
+      handleUnexpectedQueryException(operation, e, "");
+    }
+    return status;
+  }
+
   @Override
   public DeviceSummary deviceSummary(DeviceQuery deviceQuery) throws SQLException, TsdbException {
     DeviceSummary deviceSummary = null;

core/src/main/java/cn/edu/tsinghua/iot/benchmark/tsdb/IDatabase.java

@@ -180,6 +180,11 @@ default Status deviceQuery(DeviceQuery deviceQuery) throws SQLException, TsdbExc
     return new Status(false, 0, workloadException, workloadException.getMessage());
   }
 
+  default Status deviceQuery(String sql) {
+    WorkloadException workloadException = new WorkloadException("Not Supported Verification Query");
+    return new Status(false, 0, workloadException, workloadException.getMessage());
+  }
+
   /** get summary of device */
   default DeviceSummary deviceSummary(DeviceQuery deviceQuery) throws SQLException, TsdbException {
     throw new TsdbException("Not Supported get summary of device.");

iotdb-2.0/src/main/java/cn/edu/tsinghua/iot/benchmark/iotdb200/IoTDB.java

@@ -589,6 +589,21 @@ public Status deviceQuery(DeviceQuery deviceQuery) throws SQLException, TsdbExce
     return new Status(true, 0, sql, result);
   }
 
+  @Override
+  public Status deviceQuery(String sql) {
+    if (!config.isIS_QUIET_MODE()) {
+      LOGGER.info("Query: {}", sql);
+    }
+    List<List<Object>> result;
+    try {
+      result = dmlStrategy.deviceQueryImpl(sql);
+    } catch (Exception e) {
+      LOGGER.error("Query Error: {}", sql, e);
+      return new Status(false, new TsdbException("Failed to query"), "Failed to query.");
+    }
+    return new Status(true, 0, sql, result);
+  }
+
   protected String getDeviceQuerySql(
       DeviceSchema deviceSchema, long startTimeStamp, long endTimeStamp) {
     StringBuffer sql = new StringBuffer();