You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "LJW21-02 (via GitHub)" <gi...@apache.org> on 2026/01/20 10:22:34 UTC
[PR] Fix data export logic to avoid redundant high frequency query [iotdb]
LJW21-02 opened a new pull request, #17049:
URL: https://github.com/apache/iotdb/pull/17049
# Major Changes
- Fixed SQL file export logic to avoid redundant queries during high-frequency operations
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix data export logic to avoid redundant high frequency query [iotdb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17049:
URL: https://github.com/apache/iotdb/pull/17049#discussion_r2707751900
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index)) || "STRING".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
Review Comment:
The variable `currentLines` is being incremented twice in each iteration: first with `currentLines += 1` on line 184, and then again with `currentLines++` on line 189. This causes the counter to increment by 2 per row instead of 1, which will result in file splits occurring at half the intended `linesPerFile` threshold. Remove the post-increment operator from the if condition and use just `currentLines >= linesPerFile`.
```suggestion
if (currentLines >= linesPerFile) {
```
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index)) || "STRING".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
+ writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
+ }
+ }
Review Comment:
The FileWriter resource may not be closed if the loop terminates before reaching the `linesPerFile` threshold. After the while loop completes, add a check to close the writer if it's still open (i.e., if writer is not null, flush and close it). This prevents a resource leak.
```suggestion
}
if (writer != null) {
writer.flush();
writer.close();
}
```
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index)) || "STRING".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
+ writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
Review Comment:
After closing the current file and incrementing the file index, `currentLines` should be reset to 0 to properly track the line count for the new file. Without this reset, the line counting will be incorrect for subsequent files.
```suggestion
if (currentLines >= linesPerFile) {
writer.flush();
writer.close();
fileIndex += 1;
writer = null;
currentLines = 0;
```
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index)) || "STRING".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
+ writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
+ }
+ }
+ ioTPrinter.print("\n");
+ }
+
+ private void exportToSqlFileWithoutAlign(SessionDataSet sessionDataSet, String filePath)
throws IoTDBConnectionException, StatementExecutionException, IOException {
processedRows = 0;
lastPrintTime = 0;
+
List<String> headers = sessionDataSet.getColumnNames();
- int fileIndex = 0;
+ List<String> measurementNames = new ArrayList<>();
String deviceName = null;
- boolean writeNull = false;
- List<String> seriesList = new ArrayList<>(headers);
if (CollectionUtils.isEmpty(headers) || headers.size() <= 1) {
- writeNull = true;
+ return;
} else {
- if (headers.contains("Device")) {
- seriesList.remove("Time");
- seriesList.remove("Device");
- } else {
- Path path = new Path(seriesList.get(1), true);
- deviceName = path.getDeviceString();
- seriesList.remove("Time");
- for (int i = 0; i < seriesList.size(); i++) {
- String series = seriesList.get(i);
- path = new Path(series, true);
- seriesList.set(i, path.getMeasurement());
+ headers.remove("Time");
+ Path path = new Path(headers.get(0), true);
+ deviceName = path.getDeviceString();
+ for (String header : headers) {
+ path = new Path(header, true);
+ String meas = path.getMeasurement();
+ if (path.getDeviceString().equals(deviceName)) {
+ measurementNames.add(meas);
}
}
}
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ return;
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- boolean fromOuterloop = false;
+ int totalColumns = measurementNames.size();
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
while (iterator.next()) {
- fromOuterloop = true;
- final String finalFilePath = filePath + "_" + fileIndex + ".sql";
- try (FileWriter writer = new FileWriter(finalFilePath)) {
- if (writeNull) {
- break;
- }
- int i = 0;
- while (i++ < linesPerFile && (fromOuterloop || iterator.next())) {
- fromOuterloop = false;
- List<String> headersTemp = new ArrayList<>(seriesList);
- List<String> timeseries = new ArrayList<>();
- if (headers.contains("Device")) {
- deviceName = iterator.getString(2);
- if (deviceName.startsWith(SYSTEM_DATABASE + ".")
- || deviceName.startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- for (String header : headersTemp) {
- timeseries.add(deviceName + "." + header);
- }
- } else {
- if (headers.get(1).startsWith(SYSTEM_DATABASE + ".")
- || headers.get(1).startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- timeseries.addAll(headers);
- timeseries.remove(0);
- }
- long timestamp = iterator.getLong(1);
- String sqlMiddle =
- Boolean.TRUE.equals(aligned)
- ? " ALIGNED VALUES (" + timestamp + ","
- : " VALUES (" + timestamp + ",";
- List<String> values = new ArrayList<>();
- int startIndex = headers.contains("Device") ? 2 : 1;
- for (int index = startIndex; index < totalColumns; index++) {
- SessionDataSet sessionDataSet2 =
- session.executeQueryStatement(
- "SHOW TIMESERIES " + timeseries.get(index - startIndex), timeout);
- SessionDataSet.DataIterator iterator2 = sessionDataSet2.iterator();
- if (iterator2.next()) {
- if (iterator.isNull(index + 1)) {
- headersTemp.remove(seriesList.get(index - startIndex));
- continue;
- }
- String value = iterator.getString(index + 1);
- if ("TEXT".equalsIgnoreCase(iterator2.getString(4))) {
- values.add("\"" + value + "\"");
- } else {
- values.add(value);
- }
- } else {
- headersTemp.remove(seriesList.get(index - startIndex));
- }
- }
- if (CollectionUtils.isNotEmpty(headersTemp)) {
- writer.write(
- "INSERT INTO "
- + deviceName
- + "(TIMESTAMP,"
- + String.join(",", headersTemp)
- + ")"
- + sqlMiddle
- + String.join(",", values)
- + ");\n");
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- }
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 0; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index + 1)) || "STRING".equalsIgnoreCase(columnTypeList.get(index + 1))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 2)));
+ } else {
+ values.add(iterator.getString(index + 2));
}
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
Review Comment:
The variable `currentLines` is being incremented twice in each iteration: first with `currentLines += 1` on line 252, and then again with `currentLines++` on line 257. This causes the counter to increment by 2 per row instead of 1, which will result in file splits occurring at half the intended `linesPerFile` threshold. Remove the post-increment operator from the if condition and use just `currentLines >= linesPerFile`.
```suggestion
if (currentLines >= linesPerFile) {
```
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index)) || "STRING".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
+ writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
+ }
+ }
+ ioTPrinter.print("\n");
+ }
+
+ private void exportToSqlFileWithoutAlign(SessionDataSet sessionDataSet, String filePath)
throws IoTDBConnectionException, StatementExecutionException, IOException {
processedRows = 0;
lastPrintTime = 0;
+
List<String> headers = sessionDataSet.getColumnNames();
- int fileIndex = 0;
+ List<String> measurementNames = new ArrayList<>();
String deviceName = null;
- boolean writeNull = false;
- List<String> seriesList = new ArrayList<>(headers);
if (CollectionUtils.isEmpty(headers) || headers.size() <= 1) {
- writeNull = true;
+ return;
} else {
- if (headers.contains("Device")) {
- seriesList.remove("Time");
- seriesList.remove("Device");
- } else {
- Path path = new Path(seriesList.get(1), true);
- deviceName = path.getDeviceString();
- seriesList.remove("Time");
- for (int i = 0; i < seriesList.size(); i++) {
- String series = seriesList.get(i);
- path = new Path(series, true);
- seriesList.set(i, path.getMeasurement());
+ headers.remove("Time");
+ Path path = new Path(headers.get(0), true);
+ deviceName = path.getDeviceString();
+ for (String header : headers) {
+ path = new Path(header, true);
+ String meas = path.getMeasurement();
+ if (path.getDeviceString().equals(deviceName)) {
+ measurementNames.add(meas);
}
}
}
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ return;
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- boolean fromOuterloop = false;
+ int totalColumns = measurementNames.size();
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
while (iterator.next()) {
- fromOuterloop = true;
- final String finalFilePath = filePath + "_" + fileIndex + ".sql";
- try (FileWriter writer = new FileWriter(finalFilePath)) {
- if (writeNull) {
- break;
- }
- int i = 0;
- while (i++ < linesPerFile && (fromOuterloop || iterator.next())) {
- fromOuterloop = false;
- List<String> headersTemp = new ArrayList<>(seriesList);
- List<String> timeseries = new ArrayList<>();
- if (headers.contains("Device")) {
- deviceName = iterator.getString(2);
- if (deviceName.startsWith(SYSTEM_DATABASE + ".")
- || deviceName.startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- for (String header : headersTemp) {
- timeseries.add(deviceName + "." + header);
- }
- } else {
- if (headers.get(1).startsWith(SYSTEM_DATABASE + ".")
- || headers.get(1).startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- timeseries.addAll(headers);
- timeseries.remove(0);
- }
- long timestamp = iterator.getLong(1);
- String sqlMiddle =
- Boolean.TRUE.equals(aligned)
- ? " ALIGNED VALUES (" + timestamp + ","
- : " VALUES (" + timestamp + ",";
- List<String> values = new ArrayList<>();
- int startIndex = headers.contains("Device") ? 2 : 1;
- for (int index = startIndex; index < totalColumns; index++) {
- SessionDataSet sessionDataSet2 =
- session.executeQueryStatement(
- "SHOW TIMESERIES " + timeseries.get(index - startIndex), timeout);
- SessionDataSet.DataIterator iterator2 = sessionDataSet2.iterator();
- if (iterator2.next()) {
- if (iterator.isNull(index + 1)) {
- headersTemp.remove(seriesList.get(index - startIndex));
- continue;
- }
- String value = iterator.getString(index + 1);
- if ("TEXT".equalsIgnoreCase(iterator2.getString(4))) {
- values.add("\"" + value + "\"");
- } else {
- values.add(value);
- }
- } else {
- headersTemp.remove(seriesList.get(index - startIndex));
- }
- }
- if (CollectionUtils.isNotEmpty(headersTemp)) {
- writer.write(
- "INSERT INTO "
- + deviceName
- + "(TIMESTAMP,"
- + String.join(",", headersTemp)
- + ")"
- + sqlMiddle
- + String.join(",", values)
- + ");\n");
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- }
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 0; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index + 1)) || "STRING".equalsIgnoreCase(columnTypeList.get(index + 1))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 2)));
+ } else {
+ values.add(iterator.getString(index + 2));
}
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
Review Comment:
After closing the current file and incrementing the file index, `currentLines` should be reset to 0 to properly track the line count for the new file. Without this reset, the line counting will be incorrect for subsequent files.
```suggestion
if (currentLines >= linesPerFile) {
writer.flush();
writer.close();
fileIndex += 1;
writer = null;
currentLines = 0;
```
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index)) || "STRING".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
+ writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
+ }
+ }
+ ioTPrinter.print("\n");
+ }
+
+ private void exportToSqlFileWithoutAlign(SessionDataSet sessionDataSet, String filePath)
throws IoTDBConnectionException, StatementExecutionException, IOException {
processedRows = 0;
lastPrintTime = 0;
+
List<String> headers = sessionDataSet.getColumnNames();
- int fileIndex = 0;
+ List<String> measurementNames = new ArrayList<>();
String deviceName = null;
- boolean writeNull = false;
- List<String> seriesList = new ArrayList<>(headers);
if (CollectionUtils.isEmpty(headers) || headers.size() <= 1) {
- writeNull = true;
+ return;
} else {
- if (headers.contains("Device")) {
- seriesList.remove("Time");
- seriesList.remove("Device");
- } else {
- Path path = new Path(seriesList.get(1), true);
- deviceName = path.getDeviceString();
- seriesList.remove("Time");
- for (int i = 0; i < seriesList.size(); i++) {
- String series = seriesList.get(i);
- path = new Path(series, true);
- seriesList.set(i, path.getMeasurement());
+ headers.remove("Time");
+ Path path = new Path(headers.get(0), true);
+ deviceName = path.getDeviceString();
+ for (String header : headers) {
+ path = new Path(header, true);
+ String meas = path.getMeasurement();
+ if (path.getDeviceString().equals(deviceName)) {
+ measurementNames.add(meas);
}
}
}
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ return;
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- boolean fromOuterloop = false;
+ int totalColumns = measurementNames.size();
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
while (iterator.next()) {
- fromOuterloop = true;
- final String finalFilePath = filePath + "_" + fileIndex + ".sql";
- try (FileWriter writer = new FileWriter(finalFilePath)) {
- if (writeNull) {
- break;
- }
- int i = 0;
- while (i++ < linesPerFile && (fromOuterloop || iterator.next())) {
- fromOuterloop = false;
- List<String> headersTemp = new ArrayList<>(seriesList);
- List<String> timeseries = new ArrayList<>();
- if (headers.contains("Device")) {
- deviceName = iterator.getString(2);
- if (deviceName.startsWith(SYSTEM_DATABASE + ".")
- || deviceName.startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- for (String header : headersTemp) {
- timeseries.add(deviceName + "." + header);
- }
- } else {
- if (headers.get(1).startsWith(SYSTEM_DATABASE + ".")
- || headers.get(1).startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- timeseries.addAll(headers);
- timeseries.remove(0);
- }
- long timestamp = iterator.getLong(1);
- String sqlMiddle =
- Boolean.TRUE.equals(aligned)
- ? " ALIGNED VALUES (" + timestamp + ","
- : " VALUES (" + timestamp + ",";
- List<String> values = new ArrayList<>();
- int startIndex = headers.contains("Device") ? 2 : 1;
- for (int index = startIndex; index < totalColumns; index++) {
- SessionDataSet sessionDataSet2 =
- session.executeQueryStatement(
- "SHOW TIMESERIES " + timeseries.get(index - startIndex), timeout);
- SessionDataSet.DataIterator iterator2 = sessionDataSet2.iterator();
- if (iterator2.next()) {
- if (iterator.isNull(index + 1)) {
- headersTemp.remove(seriesList.get(index - startIndex));
- continue;
- }
- String value = iterator.getString(index + 1);
- if ("TEXT".equalsIgnoreCase(iterator2.getString(4))) {
- values.add("\"" + value + "\"");
- } else {
- values.add(value);
- }
- } else {
- headersTemp.remove(seriesList.get(index - startIndex));
- }
- }
- if (CollectionUtils.isNotEmpty(headersTemp)) {
- writer.write(
- "INSERT INTO "
- + deviceName
- + "(TIMESTAMP,"
- + String.join(",", headersTemp)
- + ")"
- + sqlMiddle
- + String.join(",", values)
- + ");\n");
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- }
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 0; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index + 1)) || "STRING".equalsIgnoreCase(columnTypeList.get(index + 1))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 2)));
+ } else {
+ values.add(iterator.getString(index + 2));
}
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
}
- fileIndex++;
}
Review Comment:
The FileWriter resource may not be closed if the loop terminates before reaching the `linesPerFile` threshold. After the while loop completes, add a check to close the writer if it's still open (i.e., if writer is not null, flush and close it). This prevents a resource leak.
```suggestion
}
if (writer != null) {
writer.flush();
writer.close();
}
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix data export logic to avoid redundant high frequency query [iotdb]
Posted by "CritasWang (via GitHub)" <gi...@apache.org>.
CritasWang merged PR #17049:
URL: https://github.com/apache/iotdb/pull/17049
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix data export logic to avoid redundant high frequency query [iotdb]
Posted by "CritasWang (via GitHub)" <gi...@apache.org>.
CritasWang commented on code in PR #17049:
URL: https://github.com/apache/iotdb/pull/17049#discussion_r2707722095
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index))) {
Review Comment:
TEXT or STRING
##########
iotdb-client/cli/src/main/java/org/apache/iotdb/tool/data/ExportDataTree.java:
##########
@@ -136,116 +136,145 @@ public void exportBySql(String sql, int index) {
}
}
- private void exportToSqlFile(SessionDataSet sessionDataSet, String filePath)
+ private void exportToSqlFileWithAlignDevice(SessionDataSet sessionDataSet, String filePath)
+ throws IOException, IoTDBConnectionException, StatementExecutionException {
+ processedRows = 0;
+ lastPrintTime = 0;
+
+ List<String> measurementNames = sessionDataSet.getColumnNames();
+ if (CollectionUtils.isEmpty(measurementNames) || measurementNames.size() <= 1) {
+ return;
+ } else {
+ measurementNames.remove("Time");
+ measurementNames.remove("Device");
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) ALIGNED VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
+ SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
+ List<String> columnTypeList = iterator.getColumnTypeList();
+ int totalColumns = columnTypeList.size();
+ String deviceName = null;
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
+ while (iterator.next()) {
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ deviceName = iterator.getString(2);
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ continue;
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 2; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index))) {
+ values.add(String.format("\"%s\"", iterator.getString(index + 1)));
+ } else {
+ values.add(iterator.getString(index + 1));
+ }
+ }
+ long timestamp = iterator.getLong(1);
+ writer.write(String.format(sqlPrefix, deviceName, timestamp, String.join(",", values)));
+ processedRows += 1;
+ currentLines += 1;
+ if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
+ ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
+ lastPrintTime = System.currentTimeMillis();
+ }
+ if (currentLines++ >= linesPerFile) {
+ writer.flush();
+ writer.close();
+ fileIndex += 1;
+ writer = null;
+ }
+ }
+ ioTPrinter.print("\n");
+ }
+
+ private void exportToSqlFileWithoutAlign(SessionDataSet sessionDataSet, String filePath)
throws IoTDBConnectionException, StatementExecutionException, IOException {
processedRows = 0;
lastPrintTime = 0;
+
List<String> headers = sessionDataSet.getColumnNames();
- int fileIndex = 0;
+ List<String> measurementNames = new ArrayList<>();
String deviceName = null;
- boolean writeNull = false;
- List<String> seriesList = new ArrayList<>(headers);
if (CollectionUtils.isEmpty(headers) || headers.size() <= 1) {
- writeNull = true;
+ return;
} else {
- if (headers.contains("Device")) {
- seriesList.remove("Time");
- seriesList.remove("Device");
- } else {
- Path path = new Path(seriesList.get(1), true);
- deviceName = path.getDeviceString();
- seriesList.remove("Time");
- for (int i = 0; i < seriesList.size(); i++) {
- String series = seriesList.get(i);
- path = new Path(series, true);
- seriesList.set(i, path.getMeasurement());
+ headers.remove("Time");
+ Path path = new Path(headers.get(0), true);
+ deviceName = path.getDeviceString();
+ for (String header : headers) {
+ path = new Path(header, true);
+ String meas = path.getMeasurement();
+ if (path.getDeviceString().equals(deviceName)) {
+ measurementNames.add(meas);
}
}
}
+ if (deviceName.startsWith(SYSTEM_DATABASE + ".")
+ || deviceName.startsWith(AUDIT_DATABASE + ".")) {
+ return;
+ }
+ String sqlPrefix =
+ String.format(
+ "INSERT INTO %s(TIMESTAMP,%s) VALUES (%s);\n",
+ "%s", String.join(",", measurementNames), "%d,%s");
+
SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
List<String> columnTypeList = iterator.getColumnTypeList();
- int totalColumns = columnTypeList.size();
- boolean fromOuterloop = false;
+ int totalColumns = measurementNames.size();
+ int fileIndex = 0;
+ int currentLines = 0;
+ String filePathTemplate = filePath + "_%d" + ".sql";
+ FileWriter writer = null;
while (iterator.next()) {
- fromOuterloop = true;
- final String finalFilePath = filePath + "_" + fileIndex + ".sql";
- try (FileWriter writer = new FileWriter(finalFilePath)) {
- if (writeNull) {
- break;
- }
- int i = 0;
- while (i++ < linesPerFile && (fromOuterloop || iterator.next())) {
- fromOuterloop = false;
- List<String> headersTemp = new ArrayList<>(seriesList);
- List<String> timeseries = new ArrayList<>();
- if (headers.contains("Device")) {
- deviceName = iterator.getString(2);
- if (deviceName.startsWith(SYSTEM_DATABASE + ".")
- || deviceName.startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- for (String header : headersTemp) {
- timeseries.add(deviceName + "." + header);
- }
- } else {
- if (headers.get(1).startsWith(SYSTEM_DATABASE + ".")
- || headers.get(1).startsWith(AUDIT_DATABASE + ".")) {
- continue;
- }
- timeseries.addAll(headers);
- timeseries.remove(0);
- }
- long timestamp = iterator.getLong(1);
- String sqlMiddle =
- Boolean.TRUE.equals(aligned)
- ? " ALIGNED VALUES (" + timestamp + ","
- : " VALUES (" + timestamp + ",";
- List<String> values = new ArrayList<>();
- int startIndex = headers.contains("Device") ? 2 : 1;
- for (int index = startIndex; index < totalColumns; index++) {
- SessionDataSet sessionDataSet2 =
- session.executeQueryStatement(
- "SHOW TIMESERIES " + timeseries.get(index - startIndex), timeout);
- SessionDataSet.DataIterator iterator2 = sessionDataSet2.iterator();
- if (iterator2.next()) {
- if (iterator.isNull(index + 1)) {
- headersTemp.remove(seriesList.get(index - startIndex));
- continue;
- }
- String value = iterator.getString(index + 1);
- if ("TEXT".equalsIgnoreCase(iterator2.getString(4))) {
- values.add("\"" + value + "\"");
- } else {
- values.add(value);
- }
- } else {
- headersTemp.remove(seriesList.get(index - startIndex));
- }
- }
- if (CollectionUtils.isNotEmpty(headersTemp)) {
- writer.write(
- "INSERT INTO "
- + deviceName
- + "(TIMESTAMP,"
- + String.join(",", headersTemp)
- + ")"
- + sqlMiddle
- + String.join(",", values)
- + ");\n");
- processedRows += 1;
- if (System.currentTimeMillis() - lastPrintTime > updateTimeInterval) {
- ioTPrinter.printf(Constants.PROCESSED_PROGRESS, processedRows);
- lastPrintTime = System.currentTimeMillis();
- }
- }
+ if (writer == null) {
+ writer = new FileWriter(String.format(filePathTemplate, fileIndex));
+ }
+ List<String> values = new ArrayList<>();
+ for (int index = 0; index < totalColumns; index++) {
+ if ("TEXT".equalsIgnoreCase(columnTypeList.get(index + 1))) {
Review Comment:
TEXT or STRING
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org