[GitHub] [iotdb] Ring-k commented on a change in pull request #2397: [IOTDB-1077] add insertOneDeviceRecords API in java session

[GitBox <gi...@apache.org>]

Ring-k commented on a change in pull request #2397:
URL: https://github.com/apache/iotdb/pull/2397#discussion_r551673351



##########
File path: server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
##########
@@ -2503,6 +2504,52 @@ public TsFileManagement getTsFileManagement() {
     return tsFileManagement;
   }
 
+
+  public void insert(InsertRowsOfOneDevicePlan insertRowsOfOneDevicePlan)
+      throws WriteProcessException {
+//    for (InsertRowPlan plan : insertRowsOfOneDevicePlan.getRowPlans()) {
+//      insert(plan);
+//    }

Review comment:
       It's removed. Thank you for the suggestion!

##########
File path: session/src/main/java/org/apache/iotdb/session/Session.java
##########
@@ -558,6 +561,112 @@ public void insertRecords(List<String> deviceIds, List<Long> times,
     }
   }
 
+  /**
+   * Insert multiple rows, which can reduce the overhead of network. This method is just like jdbc
+   * executeBatch, we pack some insert request in batch and send them to server. If you want improve
+   * your performance, please see insertTablet method
+   * <p>
+   * Each row is independent, which could have different deviceId, time, number of measurements
+   *
+   * @see Session#insertTablet(Tablet)
+   */
+  public void insertRecordsOfOneDevice(String deviceId, List<Long> times,
+      List<List<String>> measurementsList, List<List<TSDataType>> typesList,
+      List<List<Object>> valuesList)
+      throws IoTDBConnectionException, StatementExecutionException {
+    insertRecordsOfOneDevice(deviceId, times, measurementsList, typesList, valuesList, false);
+  }
+  /**
+   * Insert multiple rows, which can reduce the overhead of network. This method is just like jdbc
+   * executeBatch, we pack some insert request in batch and send them to server. If you want improve
+   * your performance, please see insertTablet method
+   * <p>
+   * Each row is independent, which could have different deviceId, time, number of measurements
+   *
+   * @param haveSorted  whether the times have been sorted
+   * @see Session#insertTablet(Tablet)
+   */
+  public void insertRecordsOfOneDevice(String deviceId, List<Long> times,
+      List<List<String>> measurementsList, List<List<TSDataType>> typesList,
+      List<List<Object>> valuesList, boolean haveSorted)
+      throws IoTDBConnectionException, StatementExecutionException {
+    int len = times.size();
+    if (len != measurementsList.size() || len != valuesList.size()) {
+      throw new IllegalArgumentException(
+          "deviceIds, times, measurementsList and valuesList's size should be equal");
+    }
+    TSInsertRecordsOfOneDeviceReq request = genTSInsertRecordsOfOneDeviceReq(deviceId, times, measurementsList,
+        typesList, valuesList, haveSorted);
+    try {
+      getSessionConnection(deviceId).insertRecordsOfOneDevice(request);
+    } catch (RedirectException e) {
+      handleRedirection(deviceId, e.getEndPoint());
+    }
+  }
+
+  private TSInsertRecordsOfOneDeviceReq genTSInsertRecordsOfOneDeviceReq(String deviceId, List<Long> times,
+      List<List<String>> measurementsList, List<List<TSDataType>> typesList,
+      List<List<Object>> valuesList, boolean haveSorted) throws IoTDBConnectionException, BatchExecutionException {
+    // check params size
+    int len = times.size();
+    if (len != measurementsList.size() || len != valuesList.size()) {
+      throw new IllegalArgumentException(
+          "times, measurementsList and valuesList's size should be equal");
+    }
+
+    if (haveSorted) {
+      if (!checkSorted(times)) {
+        throw new BatchExecutionException("Times in InsertOneDeviceRecords are not in ascending order");
+      }
+    } else {
+      //sort
+      Integer[] index = new Integer[times.size()];
+      Integer[] index2 = new Integer[times.size()];
+      for (int i = 0; i < times.size(); i++) {
+        index2[i] = index[i] = i;
+      }
+      Arrays.sort(index, Comparator.comparingLong(times::get));
+      Arrays.sort(index2, Comparator.comparingInt(x -> index[x]));
+      times.sort(Long::compareTo);
+      //sort measurementList
+      measurementsList = sortList(measurementsList, index2);
+      //sort typesList
+      typesList = sortList(typesList, index2);
+      //sort values
+      valuesList = sortList(valuesList, index2);
+    }
+
+    TSInsertRecordsOfOneDeviceReq request = new TSInsertRecordsOfOneDeviceReq();
+    request.setDeviceId(deviceId);
+    request.setTimestamps(times);
+    request.setMeasurementsList(measurementsList);
+    List<ByteBuffer> buffersList = objectValuesListToByteBufferList(valuesList, typesList);
+    request.setValuesList(buffersList);
+    return request;
+  }
+
+  @SuppressWarnings("squid:S3740")
+  private List sortList(List source, Integer[] index) {
+    Object[] result = new Object[source.size()];
+    for (int i = 0; i < index.length; i++) {
+      result[index[i]] = source.get(i);
+    }
+    return Arrays.asList(result);
+  }
+
+  private List<ByteBuffer> objectValuesListToByteBufferList(List<List<Object>> valuesList, List<List<TSDataType>> typesList)
+      throws IoTDBConnectionException {
+    List<ByteBuffer> buffersList = new ArrayList<>();
+    for (int i = 0; i < valuesList.size(); i++) {
+      ByteBuffer buffer = ByteBuffer.allocate(calculateLength(typesList.get(i), valuesList.get(i)));
+      putValues(typesList.get(i), valuesList.get(i), buffer);
+//      buffer.flip();

Review comment:
       It's removed. Thank you for the suggestion!




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org