jt2594838 commented on a change in pull request #31: Fix sonar
URL: https://github.com/apache/incubator-iotdb/pull/31#discussion_r251008757
##########
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 + ")";
Review comment:
This can be extracted as a useful function that transforms a record into an
insertion sql.
----------------------------------------------------------------
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
