jt2594838 commented on a change in pull request #31: Fix sonar
URL: https://github.com/apache/incubator-iotdb/pull/31#discussion_r251010380
##########
File path:
iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java
##########
@@ -320,318 +331,298 @@ public void afterReceiving() {
*/
@Override
public void getFileNodeInfo() throws TException {
- // String filePath = postbackPath + uuid.get() + File.separator + "data";
- // File root = new File(filePath);
- // File[] files = root.listFiles();
- // int num = 0;
- // for (File storageGroupPB : files) {
- // List<String> filesPath = new ArrayList<>();
- // File[] filesSG = storageGroupPB.listFiles();
- // for (File fileTF : filesSG) { // fileTF means TsFiles
- // Map<String, Long> startTimeMap = new HashMap<>();
- // Map<String, Long> endTimeMap = new HashMap<>();
- // TsRandomAccessLocalFileReader input = null;
- // try {
- // input = new TsRandomAccessLocalFileReader(fileTF.getAbsolutePath());
- // FileReader reader = new FileReader(input);
- // Map<String, TsDevice> deviceIdMap =
reader.getFileMetaData().getDeviceMap();
- // Iterator<String> it = deviceIdMap.keySet().iterator();
- // while (it.hasNext()) {
- // String key = it.next().toString(); // key represent device
- // TsDevice deltaObj = deviceIdMap.get(key);
- // startTimeMap.put(key, deltaObj.startTime);
- // endTimeMap.put(key, deltaObj.endTime);
- // }
- // } catch (Exception e) {
- // LOGGER.error("IoTDB post back receiver: unable to read tsfile {}
because {}",
- // fileTF.getAbsolutePath(), e.getMessage());
- // } finally {
- // try {
- // input.close();
- // } catch (IOException e) {
- // LOGGER.error("IoTDB receiver : Cannot close file stream {} because {}",
- // fileTF.getAbsolutePath(), e.getMessage());
- // }
- // }
- // fileNodeStartTime.get().put(fileTF.getAbsolutePath(), startTimeMap);
- // fileNodeEndTime.get().put(fileTF.getAbsolutePath(), endTimeMap);
- // filesPath.add(fileTF.getAbsolutePath());
- // num++;
- // LOGGER.info("IoTDB receiver : Getting FileNode Info has complete : " +
num + "/" +
- // fileNum.get());
- // }
- // fileNodeMap.get().put(storageGroupPB.getName(), filesPath);
- // }
+ String filePath = postbackPath + uuid.get() + File.separator + "data";
+ File root = new File(filePath);
+ File[] files = root.listFiles();
+ int num = 0;
+ for (File storageGroupPB : files) {
+ List<String> filesPath = new ArrayList<>();
+ File[] filesSG = storageGroupPB.listFiles();
+ for (File fileTF : filesSG) { // fileTF means TsFiles
+ Map<String, Long> startTimeMap = new HashMap<>();
+ Map<String, Long> endTimeMap = new HashMap<>();
+ TsFileSequenceReader reader = null;
+ try {
+ reader = new TsFileSequenceReader(fileTF.getAbsolutePath());
+ Map<String, TsDeviceMetadataIndex> deviceIdMap =
reader.readFileMetadata().getDeviceMap();
+ Iterator<String> it = deviceIdMap.keySet().iterator();
+ while (it.hasNext()) {
+ String key = it.next(); // key represent device
+ TsDeviceMetadataIndex device = deviceIdMap.get(key);
+ startTimeMap.put(key, device.getStartTime());
+ endTimeMap.put(key, device.getEndTime());
+ }
+ } catch (Exception e) {
+ LOGGER.error("IoTDB post back receiver: unable to read tsfile {}
because {}",
+ fileTF.getAbsolutePath(), e.getMessage());
+ } finally {
+ try {
+ if (reader != null) {
+ reader.close();
+ }
+ } catch (IOException e) {
+ LOGGER.error("IoTDB receiver : Cannot close file stream {} because
{}",
+ fileTF.getAbsolutePath(), e.getMessage());
+ }
+ }
+ fileNodeStartTime.get().put(fileTF.getAbsolutePath(), startTimeMap);
+ fileNodeEndTime.get().put(fileTF.getAbsolutePath(), endTimeMap);
+ filesPath.add(fileTF.getAbsolutePath());
+ num++;
+ LOGGER.info(String
+ .format("IoTDB receiver : Getting FileNode Info has complete :
%d/%d", num,
+ fileNum.get()));
+ }
+ fileNodeMap.get().put(storageGroupPB.getName(), filesPath);
+ }
}
/**
* Insert all data in the tsfile into IoTDB.
*/
@Override
public void mergeOldData(String filePath) throws TException {
- // Set<String> timeseries = new HashSet<>();
- // TsRandomAccessLocalFileReader input = null;
- // Connection connection = null;
- // Statement statement = null;
- // try {
- // Class.forName(JDBC_DRIVER_NAME);
- // connection = DriverManager.getConnection("jdbc:iotdb://localhost:" +
- // tsfileDBConfig.rpcPort + "/", "root",
- // "root");
- // statement = connection.createStatement();
- // int count = 0;
- //
- // input = new TsRandomAccessLocalFileReader(filePath);
- // FileReader reader = new FileReader(input);
- // Map<String, TsdeviceId> deviceIdMap =
reader.getFileMetaData().getDeviceMap();
- // Iterator<String> it = deviceIdMap.keySet().iterator();
- // while (it.hasNext()) {
- // String key = it.next().toString(); // key represent devices
- // TsdeviceId deltaObj = deviceIdMap.get(key);
- // TsRowGroupBlockMetaData blockMeta = new TsRowGroupBlockMetaData();
- //
blockMeta.convertToTSF(ReadWriteThriftFormatUtils.readRowGroupBlockMetaData(input,
- // deltaObj.offset,
- // deltaObj.metadataBlockSize));
- // List<RowGroupMetaData> rowGroupMetadataList = blockMeta.getRowGroups();
- // timeseries.clear();
- // // firstly, get all timeseries in the same device
- // for (RowGroupMetaData rowGroupMetaData : rowGroupMetadataList) {
- // List<TimeSeriesChunkMetaData> timeSeriesChunkMetaDataList =
rowGroupMetaData
- // .getTimeSeriesChunkMetaDataList();
- // for (TimeSeriesChunkMetaData timeSeriesChunkMetaData :
timeSeriesChunkMetaDataList) {
- // TimeSeriesChunkProperties properties =
timeSeriesChunkMetaData.getProperties();
- // String measurementUID = properties.getMeasurementUID();
- // measurementUID = key + "." + measurementUID;
- // timeseries.add(measurementUID);
- // }
- // }
- // // secondly, use tsFile Reader to form SQL
- //
- // TsFile readTsFile = new TsFile(input);
- // ArrayList<Path> paths = new ArrayList<>();
- // paths.clear();
- // for (String timesery : timeseries) {
- // paths.add(new Path(timesery));
- // }
- // OnePassQueryDataSet queryDataSet = readTsFile.query(paths, null, null);
- // while (queryDataSet.hasNextRecord()) {
- // OldRowRecord record = queryDataSet.getNextRecord();
- // List<Field> fields = record.getFields();
- // String sql_front = null;
- // for (Field field : fields) {
- // if (field.toString() != "null") {
- // sql_front = "insert into " + field.deviceId + "(timestamp";
- // break;
- // }
- // }
- // String sql_rear = ") values(" + record.timestamp;
- // for (Field field : fields) {
- // if (field.toString() != "null") {
- // sql_front = sql_front + "," + field.measurementId.toString();
- // if (field.dataType == TSDataType.TEXT) {
- // sql_rear = sql_rear + "," + "'" + field.toString() + "'";
- // } else {
- // sql_rear = sql_rear + "," + field.toString();
- // }
- // }
- // }
- // String sql = sql_front + sql_rear + ")";
- //
- // statement.addBatch(sql);
- // count++;
- // if (count > 10000) {
- // statement.executeBatch();
- // statement.clearBatch();
- // count = 0;
- // }
- // }
- // }
- // statement.executeBatch();
- // statement.clearBatch();
- // } catch (IOException e) {
- // LOGGER.error("IoTDB receiver can not parse tsfile into SQL because{}",
e.getMessage());
- // } catch (SQLException | ClassNotFoundException e) {
- // LOGGER.error("IoTDB post back receiver: jdbc cannot connect to IoTDB
because {}",
- // e.getMessage());
- // } finally {
- // try {
- // input.close();
- // } catch (IOException e) {
- // LOGGER.error("IoTDB receiver : Cannot close file stream {} because {}",
filePath);
- // }
- // try {
- // if (statement != null)
- // statement.close();
- // if (connection != null)
- // connection.close();
- // } catch (SQLException e) {
- // LOGGER.error("IoTDB receiver : Can not close JDBC connection because
{}", e.getMessage());
- // }
- // }
+ Set<String> timeseries = new HashSet<>();
+ Statement statement = null;
+ TsFileSequenceReader reader = null;
+ try (Connection connection = DriverManager.getConnection(
+ String.format("jdbc:iotdb://localhost:%d/", config.rpcPort), "root",
+ "root")) {
+ Class.forName(JDBC_DRIVER_NAME);
+ statement = connection.createStatement();
+ int count = 0;
+
+ reader = new TsFileSequenceReader(filePath);
+ Map<String, TsDeviceMetadataIndex> deviceIdMap =
reader.readFileMetadata().getDeviceMap();
+ Iterator<String> it = deviceIdMap.keySet().iterator();
+ while (it.hasNext()) {
+ String deviceId = it.next(); // deviceId represent devices
+ TsDeviceMetadataIndex deviceMetadataIndex = deviceIdMap.get(deviceId);
+ TsDeviceMetadata deviceMetadata =
reader.readTsDeviceMetaData(deviceMetadataIndex);
+ List<ChunkGroupMetaData> rowGroupMetadataList =
deviceMetadata.getChunkGroupMetaDataList();
+ timeseries.clear();
+ // firstly, get all timeseries in the same device
+ for (ChunkGroupMetaData chunkGroupMetaData : rowGroupMetadataList) {
+ List<ChunkMetaData> chunkMetaDataList = chunkGroupMetaData
+ .getChunkMetaDataList();
+ for (ChunkMetaData chunkMetaData : chunkMetaDataList) {
+ String measurementUID = chunkMetaData.getMeasurementUid();
+ measurementUID = deviceId + "." + measurementUID;
+ timeseries.add(measurementUID);
+ }
+ }
+ // secondly, use tsFile Reader to form SQL
+
+ ReadOnlyTsFile readTsFile = new ReadOnlyTsFile(reader);
+ List<Path> paths = new ArrayList<>();
+ paths.clear();
+ for (String timesery : timeseries) {
+ paths.add(new Path(timesery));
+ }
+ QueryExpression queryExpression = QueryExpression.create(paths, null);
+ QueryDataSet queryDataSet = readTsFile.query(queryExpression);
+ while (queryDataSet.hasNext()) {
+ RowRecord record = queryDataSet.next();
+ List<Field> fields = record.getFields();
+ String sqlFront = String.format("insert into %s(timestamp",
deviceId);
+ String sqlRear = String.format(") values(%d", record.getTimestamp());
+ for (int i = 0; i < fields.size(); i++) {
+ Field field = fields.get(i);
+ if (field.toString() != "null") {
+ sqlFront = String.format("%s,%s", sqlFront,
paths.get(i).getMeasurement());
+ if (fields.get(i).getDataType() == TSDataType.TEXT) {
+ sqlRear = String.format("%s,'%s'", sqlRear, field.toString());
+ } else {
+ sqlRear = String.format("%s,%s", sqlRear, field.toString());
+ }
+ }
+ }
+ String sql = sqlFront + sqlRear + ")";
+
+ statement.addBatch(sql);
+ count++;
+ if (count > 10000) {
+ statement.executeBatch();
+ statement.clearBatch();
+ count = 0;
+ }
+ }
+ }
+ statement.executeBatch();
+ statement.clearBatch();
+ } catch (IOException e) {
+ LOGGER.error("IoTDB receiver can not parse tsfile into SQL because{}",
e.getMessage());
+ } catch (SQLException | ClassNotFoundException e) {
+ LOGGER.error("IoTDB post back receiver: jdbc cannot connect to IoTDB
because {}",
+ e.getMessage());
+ } finally {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ LOGGER.error("IoTDB receiver : Cannot close file stream {} because
{}", filePath,
+ e.getMessage());
+ }
+ try {
+ if (statement != null) {
+ statement.close();
+ }
+ } catch (SQLException e) {
+ LOGGER.error("IoTDB receiver : Can not close JDBC connection because
{}", e.getMessage());
+ }
+ }
}
/**
* Insert those valid data in the tsfile into IoTDB
*
* @param overlapFiles:files which are conflict with the postback file
*/
- public void mergeOldData(String filePath, List<String> overlapFiles) throws
TException {
- // Set<String> timeseries = new HashSet<>();
- // TsRandomAccessLocalFileReader input = null;
- // Connection connection = null;
- // Statement statement = null;
- // try {
- // Class.forName(JDBC_DRIVER_NAME);
- // connection = DriverManager.getConnection("jdbc:iotdb://localhost:" +
- // tsfileDBConfig.rpcPort + "/", "root",
- // "root");
- // statement = connection.createStatement();
- // int count = 0;
- //
- // input = new TsRandomAccessLocalFileReader(filePath);
- // FileReader reader = new FileReader(input);
- // Map<String, TsdeviceId> deviceIdMap =
reader.getFileMetaData().getDeviceMap();
- // Iterator<String> it = deviceIdMap.keySet().iterator();
- // while (it.hasNext()) {
- // String key = it.next().toString(); // key represent devices
- // TsdeviceId deltaObj = deviceIdMap.get(key);
- // TsRowGroupBlockMetaData blockMeta = new TsRowGroupBlockMetaData();
- //
blockMeta.convertToTSF(ReadWriteThriftFormatUtils.readRowGroupBlockMetaData(input,
- // deltaObj.offset,
- // deltaObj.metadataBlockSize));
- // List<RowGroupMetaData> rowGroupMetadataList = blockMeta.getRowGroups();
- // timeseries.clear();
- // // firstly, get all timeseries in the same device
- // for (RowGroupMetaData rowGroupMetaData : rowGroupMetadataList) {
- // List<TimeSeriesChunkMetaData> timeSeriesChunkMetaDataList =
rowGroupMetaData
- // .getTimeSeriesChunkMetaDataList();
- // for (TimeSeriesChunkMetaData timeSeriesChunkMetaData :
timeSeriesChunkMetaDataList) {
- // TimeSeriesChunkProperties properties =
timeSeriesChunkMetaData.getProperties();
- // String measurementUID = properties.getMeasurementUID();
- // measurementUID = key + "." + measurementUID;
- // timeseries.add(measurementUID);
- // }
- // }
- // // secondly, use tsFile Reader to form SQL
- //
- // TsFile readTsFile = new TsFile(input);
- // ArrayList<Path> paths = new ArrayList<>();
- // for (String timesery : timeseries) { // compare data with one timesery
in a round to get
- // valid data
- // paths.clear();
- // paths.add(new Path(timesery));
- // Map<String, String> originDataPoint = new HashMap<>();
- // Map<String, String> newDataPoint = new HashMap<>();
- // String sqlFormat = "insert into %s(timestamp,%s) values(%s,%s)";
- // OnePassQueryDataSet queryDataSet = readTsFile.query(paths, null, null);
- // while (queryDataSet.hasNextRecord()) {
- // OldRowRecord record = queryDataSet.getNextRecord();
- // List<Field> fields = record.getFields();
- // String sql;
- // for (Field field : fields) { // get all data with the timesery in the
postback file
- // if (field.toString() != "null") {
- // sql = String.format(sqlFormat, field.deviceId,
field.measurementId.toString(),
- // record.timestamp, "%s");
- // if (field.dataType == TSDataType.TEXT) {
- // newDataPoint.put(sql, "'" + field.toString() + "'");
- // } else {
- // newDataPoint.put(sql, field.toString());
- // }
- //
- // }
- // }
- // }
- // for (String overlapFile : overlapFiles) // get all data with the
timesery in all overlap
- // files.
- // {
- // TsRandomAccessLocalFileReader inputOverlap = null;
- // try {
- // inputOverlap = new TsRandomAccessLocalFileReader(overlapFile);
- // TsFile readTsFileOverlap = new TsFile(inputOverlap);
- // OnePassQueryDataSet queryDataSetOverlap =
readTsFileOverlap.query(paths, null, null);
- // while (queryDataSetOverlap.hasNextRecord()) {
- // OldRowRecord recordOverlap = queryDataSetOverlap.getNextRecord();
- // List<Field> fields = recordOverlap.getFields();
- // String sql;
- // for (Field field : fields) {
- // if (field.toString() != "null") {
- // sql = String.format(sqlFormat, field.deviceId,
- // field.measurementId.toString(), recordOverlap.timestamp, "%s");
- // if (field.dataType == TSDataType.TEXT) {
- // originDataPoint.put(sql, "'" + field.toString() + "'");
- // } else {
- // originDataPoint.put(sql, field.toString());
- // }
- // }
- // }
- // }
- // } catch (IOException e) {
- // LOGGER.error("IoTDB post back receiver: unable to read tsfile {}
because {}",
- // overlapFile,
- // e.getMessage());
- // } finally {
- // try {
- // inputOverlap.close();
- // } catch (IOException e) {
- // LOGGER.error("IoTDB receiver : Cannot close file stream {} because {}",
overlapFile,
- // e.getMessage());
- // }
- // }
- // }
- // if (originDataPoint.isEmpty()) { // If there has no overlap data with
the timesery,
- // inserting all
- // // data in the postback file
- // for (Entry<String, String> entry : newDataPoint.entrySet()) {
- // String sql = String.format(entry.getKey(), entry.getValue());
- // statement.addBatch(sql);
- // count++;
- // if (count > 10000) {
- // statement.executeBatch();
- // statement.clearBatch();
- // count = 0;
- // }
- // }
- // } else { // Compare every data to get valid data
- // for (Entry<String, String> entry : newDataPoint.entrySet()) {
- // if (!originDataPoint.containsKey(entry.getKey())
- // || (originDataPoint.containsKey(entry.getKey())
- // && !originDataPoint.get(entry.getKey()).equals(entry.getValue()))) {
- // String sql = String.format(entry.getKey(), entry.getValue());
- // statement.addBatch(sql);
- // count++;
- // if (count > 10000) {
- // statement.executeBatch();
- // statement.clearBatch();
- // count = 0;
- // }
- // }
- // }
- // }
- // }
- // }
- // statement.executeBatch();
- // statement.clearBatch();
- // } catch (IOException e) {
- // LOGGER.error("IoTDB receiver can not parse tsfile into SQL because{}",
e.getMessage());
- // } catch (SQLException | ClassNotFoundException e) {
- // LOGGER.error("IoTDB post back receiver: jdbc cannot connect to IoTDB
because {}",
- // e.getMessage());
- // } finally {
- // try {
- // input.close();
- // } catch (IOException e) {
- // LOGGER.error("IoTDB receiver : Cannot close file stream {} because {}",
filePath,
- // e.getMessage());
- // }
- // try {
- // if (statement != null)
- // statement.close();
- // if (connection != null)
- // connection.close();
- // } catch (SQLException e) {
- // LOGGER.error("IoTDB receiver : Can not close JDBC connection because
{}", e.getMessage());
- // }
- // }
+ public void mergeOldData(String filePath, List<String> overlapFiles) {
+ Set<String> timeseries = new HashSet<>();
+ TsFileSequenceReader reader = null;
+ Statement statement = null;
+ try (Connection connection = DriverManager.getConnection(
+ String.format("jdbc:iotdb://localhost:%d/", config.rpcPort), "root",
+ "root")) {
+ Class.forName(JDBC_DRIVER_NAME);
+ statement = connection.createStatement();
+ int count = 0;
+
+ reader = new TsFileSequenceReader(filePath);
+ Map<String, TsDeviceMetadataIndex> deviceIdMap =
reader.readFileMetadata().getDeviceMap();
+ Iterator<String> it = deviceIdMap.keySet().iterator();
+ while (it.hasNext()) {
+ String key = it.next(); // key represent devices
+ TsDeviceMetadataIndex deviceMetadataIndex = deviceIdMap.get(key);
+ TsDeviceMetadata deviceMetadata =
reader.readTsDeviceMetaData(deviceMetadataIndex);
+ List<ChunkGroupMetaData> chunkGroupMetaDataList = deviceMetadata
+ .getChunkGroupMetaDataList();
+ timeseries.clear();
+ // firstly, get all timeseries in the same device
+ for (ChunkGroupMetaData chunkGroupMetaData : chunkGroupMetaDataList) {
+ List<ChunkMetaData> chunkMetaDataList =
chunkGroupMetaData.getChunkMetaDataList();
+ for (ChunkMetaData timeSeriesChunkMetaData : chunkMetaDataList) {
+ String measurementUID =
timeSeriesChunkMetaData.getMeasurementUid();
+ measurementUID = key + "." + measurementUID;
+ timeseries.add(measurementUID);
+ }
+ }
+ // secondly, use tsFile Reader to form SQL
+
+ ReadOnlyTsFile readOnlyTsFile = new ReadOnlyTsFile(reader);
+ ArrayList<Path> paths = new ArrayList<>();
+ for (String timesery : timeseries) { // compare data with one timesery
in a round to get valid data
+ paths.clear();
+ paths.add(new Path(timesery));
+ Map<String, String> originDataPoint = new HashMap<>();
+ Map<String, String> newDataPoint = new HashMap<>();
+ String sqlFormat = "insert into %s(timestamp,%s) values(%s,%s)";
+ QueryExpression queryExpression = QueryExpression.create(paths,
null);
+ QueryDataSet queryDataSet = readOnlyTsFile.query(queryExpression);
+ while (queryDataSet.hasNext()) {
+ RowRecord record = queryDataSet.next();
+ List<Field> fields = record.getFields();
+ String sql;
+ for (int i = 0; i < fields.size();
+ i++) { // get all data with the timesery in the postback file
+ Field field = fields.get(i);
+ if (!field.isNull()) {
+ sql = String
+ .format(sqlFormat, key, paths.get(i).getMeasurement(),
record.getTimestamp(),
+ "%s");
+ if (field.getDataType() == TSDataType.TEXT) {
+ newDataPoint.put(sql, "'" + field.toString() + "'");
+ } else {
+ newDataPoint.put(sql, field.toString());
+ }
+ }
+ }
+ }
+ for (String overlapFile : overlapFiles) // get all data with the
timesery in all overlap files.
+ {
+ TsFileSequenceReader inputOverlap = null;
+ try {
+ inputOverlap = new TsFileSequenceReader(overlapFile);
+ ReadOnlyTsFile readTsFileOverlap = new
ReadOnlyTsFile(inputOverlap);
+ QueryDataSet queryDataSetOverlap =
readTsFileOverlap.query(queryExpression);
+ while (queryDataSetOverlap.hasNext()) {
+ RowRecord recordOverlap = queryDataSetOverlap.next();
+ List<Field> fields = recordOverlap.getFields();
+ String sql;
+ for (int i = 0; i < fields.size(); i++) {
+ Field field = fields.get(i);
+ if (!field.isNull()) {
+ sql = String.format(sqlFormat, key,
+ paths.get(i).getMeasurement(),
recordOverlap.getTimestamp(), "%s");
+ if (field.getDataType() == TSDataType.TEXT) {
+ originDataPoint.put(sql, "'" + field.toString() + "'");
+ } else {
+ originDataPoint.put(sql, field.toString());
+ }
Review comment:
The depth of nested block is too large. Try simplifying it if you can.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
