Merge pull request thulab#492 from YangYumings/addFixedSQL

Enable fixed query mode (default: true)

Author: jt2594838

configuration/conf/config.properties

@@ -336,6 +336,9 @@
 # 最长等待读时间，单位毫秒，即如果整个读操作在指定时间内没有返回，则终止此操作
 # READ_OPERATION_TIMEOUT_MS=300000
 
+# 是否启用固定查询模式,所有查询线程使用相同的设备和传感器组合，生成一致的查询 SQL
+# ENABLE_FIXED_QUERY=true
+
 ################## 操作信息：写入参数 ###################
 # 每批每个设备写入数据行数，一行是某个设备所有传感器在某一时间戳的数据
 # 每个Batch写入数据总点数=DEVICE_NUM_PER_WRITE * SENSOR_NUMBER * BATCH_SIZE_PER_WRITE
@@ -398,7 +401,7 @@
 
 # 时间过滤条件的时间起点变化步长，单位：POINT_STEP。若设为0则每个查询的时间过滤条件是一样的。
 # 计算方式为 时间间隔 = STEP_SIZE * POINT_STEP。
-# STEP_SIZE=1
+# STEP_SIZE=0
 
 # 每条查询语句中查询涉及到的传感器数量
 # QUERY_SENSOR_NUM=1

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

@@ -351,6 +351,8 @@ public class Config {
    */
   private String OPERATION_PROPORTION = "1:0:0:0:0:0:0:0:0:0:0:0";
 
+  private boolean ENABLE_FIXED_QUERY = true;
+
   private final int OPERATION_PROPORTION_LEN = 12;
   /** The number of sensors involved in each query */
   private int QUERY_SENSOR_NUM = 1;
@@ -1308,6 +1310,14 @@ public int getOPERATION_PROPORTION_LEN() {
     return this.OPERATION_PROPORTION_LEN;
   }
 
+  public boolean isENABLE_FIXED_QUERY() {
+    return ENABLE_FIXED_QUERY;
+  }
+
+  public void setENABLE_FIXED_QUERY(boolean ENABLE_FIXED_QUERY) {
+    this.ENABLE_FIXED_QUERY = ENABLE_FIXED_QUERY;
+  }
+
   public int getQUERY_SENSOR_NUM() {
     return QUERY_SENSOR_NUM;
   }

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

@@ -454,6 +454,9 @@ private void loadProps() {
             Long.parseLong(properties.getProperty("STEP_SIZE", config.getSTEP_SIZE() + "")));
         config.setOPERATION_PROPORTION(
             properties.getProperty("OPERATION_PROPORTION", config.getOPERATION_PROPORTION()));
+        config.setENABLE_FIXED_QUERY(
+            Boolean.parseBoolean(
+                properties.getProperty("ENABLE_FIXED_QUERY", config.isENABLE_FIXED_QUERY() + "")));
         config.setQUERY_SENSOR_NUM(
             Integer.parseInt(
                 properties.getProperty("QUERY_SENSOR_NUM", config.getQUERY_SENSOR_NUM() + "")));

core/src/main/java/cn/edu/tsinghua/iot/benchmark/workload/GenerateQueryWorkLoad.java

@@ -52,6 +52,9 @@ public class GenerateQueryWorkLoad extends QueryWorkLoad {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(GenerateQueryWorkLoad.class);
 
+  private static volatile List<DeviceSchema> cachedQueryDeviceSchemaListTypeAllow = null;
+  private static volatile List<DeviceSchema> cachedQueryDeviceSchemaListTypeFiltered = null;
+
   private final Random queryDeviceRandom;
   private final Random querySensorRandom;
   private static final long timeStampConst =
@@ -73,22 +76,48 @@ public GenerateQueryWorkLoad(int id) {
 
   @Override
   public PreciseQuery getPreciseQuery() throws WorkloadException {
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(true);
+    List<DeviceSchema> queryDevices = getQueryDeviceSchema(true, config.isENABLE_FIXED_QUERY());
     long timestamp = getQueryStartTimestamp(Operation.PRECISE_QUERY);
     return new PreciseQuery(queryDevices, timestamp);
   }
 
+  public List<DeviceSchema> getQueryDeviceSchema(boolean typeAllow, boolean fixedSQl)
+      throws WorkloadException {
+    if (!fixedSQl) {
+      return getQueryDeviceSchemaList(typeAllow);
+    }
+    if (typeAllow) {
+      if (cachedQueryDeviceSchemaListTypeAllow == null) {
+        synchronized (GenerateQueryWorkLoad.class) {
+          if (cachedQueryDeviceSchemaListTypeAllow == null) {
+            cachedQueryDeviceSchemaListTypeAllow = getQueryDeviceSchemaList(typeAllow);
+          }
+        }
+      }
+      return cachedQueryDeviceSchemaListTypeAllow;
+    } else {
+      if (cachedQueryDeviceSchemaListTypeFiltered == null) {
+        synchronized (GenerateQueryWorkLoad.class) {
+          if (cachedQueryDeviceSchemaListTypeFiltered == null) {
+            cachedQueryDeviceSchemaListTypeFiltered = getQueryDeviceSchemaList(typeAllow);
+          }
+        }
+      }
+      return cachedQueryDeviceSchemaListTypeFiltered;
+    }
+  }
+
   @Override
   public RangeQuery getRangeQuery() throws WorkloadException {
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(true);
+    List<DeviceSchema> queryDevices = getQueryDeviceSchema(true, config.isENABLE_FIXED_QUERY());
     long startTimestamp = getQueryStartTimestamp(Operation.RANGE_QUERY);
     long endTimestamp = startTimestamp + config.getQUERY_INTERVAL();
     return new RangeQuery(queryDevices, startTimestamp, endTimestamp);
   }
 
   @Override
   public ValueRangeQuery getValueRangeQuery() throws WorkloadException {
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(false);
+    List<DeviceSchema> queryDevices = getQueryDeviceSchema(false, config.isENABLE_FIXED_QUERY());
     long startTimestamp = getQueryStartTimestamp(Operation.VALUE_RANGE_QUERY);
     long endTimestamp = startTimestamp + config.getQUERY_INTERVAL();
     return new ValueRangeQuery(
@@ -98,7 +127,8 @@ public ValueRangeQuery getValueRangeQuery() throws WorkloadException {
   @Override
   public AggRangeQuery getAggRangeQuery() throws WorkloadException {
     List<DeviceSchema> queryDevices =
-        getQueryDeviceSchemaList(config.getQUERY_AGGREGATE_FUN().startsWith("count"));
+        getQueryDeviceSchema(
+            config.getQUERY_AGGREGATE_FUN().startsWith("count"), config.isENABLE_FIXED_QUERY());
     long startTimestamp = getQueryStartTimestamp(Operation.AGG_RANGE_QUERY);
     long endTimestamp = startTimestamp + config.getQUERY_INTERVAL();
     return new AggRangeQuery(
@@ -107,14 +137,14 @@ public AggRangeQuery getAggRangeQuery() throws WorkloadException {
 
   @Override
   public AggValueQuery getAggValueQuery() throws WorkloadException {
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(false);
+    List<DeviceSchema> queryDevices = getQueryDeviceSchema(false, config.isENABLE_FIXED_QUERY());
     return new AggValueQuery(
         queryDevices, config.getQUERY_AGGREGATE_FUN(), config.getQUERY_LOWER_VALUE());
   }
 
   @Override
   public AggRangeValueQuery getAggRangeValueQuery() throws WorkloadException {
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(false);
+    List<DeviceSchema> queryDevices = getQueryDeviceSchema(false, config.isENABLE_FIXED_QUERY());
     long startTimestamp = getQueryStartTimestamp(Operation.AGG_RANGE_VALUE_QUERY);
     long endTimestamp = startTimestamp + config.getQUERY_INTERVAL();
     return new AggRangeValueQuery(
@@ -136,7 +166,8 @@ public GroupByQuery getGroupByQuery() throws WorkloadException {
             && config.getANOTHER_DBConfig().getDB_SWITCH() == DBSwitch.DB_INFLUX_2)) {
       typeAllow = false;
     }
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(typeAllow);
+    List<DeviceSchema> queryDevices =
+        getQueryDeviceSchema(typeAllow, config.isENABLE_FIXED_QUERY());
     long startTimestamp = getQueryStartTimestamp(Operation.GROUP_BY_QUERY);
     long endTimestamp = startTimestamp + config.getQUERY_INTERVAL();
     return new GroupByQuery(
@@ -149,7 +180,7 @@ public GroupByQuery getGroupByQuery() throws WorkloadException {
 
   @Override
   public LatestPointQuery getLatestPointQuery() throws WorkloadException {
-    List<DeviceSchema> queryDevices = getQueryDeviceSchemaList(true);
+    List<DeviceSchema> queryDevices = getQueryDeviceSchema(true, config.isENABLE_FIXED_QUERY());
     long startTimestamp = getQueryStartTimestamp(Operation.LATEST_POINT_QUERY);
     long endTimestamp = startTimestamp + config.getQUERY_INTERVAL();
     return new LatestPointQuery(
@@ -188,7 +219,10 @@ private long getQueryStartTimestamp(Operation operation) {
       }
     }
     long currentQueryLoop = operationLoops.get(operation).getAndIncrement();
-    long timestampOffset = currentQueryLoop * config.getSTEP_SIZE() * config.getPOINT_STEP();
+    long timestampOffset = 0;
+    if (!config.isENABLE_FIXED_QUERY()) {
+      timestampOffset = currentQueryLoop * config.getSTEP_SIZE() * config.getPOINT_STEP();
+    }
     return Constants.START_TIMESTAMP * timeStampConst + timestampOffset;
   }