This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8ea5ed6 [IOTDB-2505] Remove tsfile in memory before delete it in
compaction (#4994)
8ea5ed6 is described below
commit 8ea5ed61c0e32815c57c85f864b66496090f5f8d
Author: Liu Xuxin <[email protected]>
AuthorDate: Sun Jan 30 09:56:44 2022 +0800
[IOTDB-2505] Remove tsfile in memory before delete it in compaction (#4994)
---
.../db/integration/IoTDBNewTsFileCompactionIT.java | 233 +++++++++++----------
.../InnerSpaceCompactionExceptionHandler.java | 118 ++++++-----
.../SizeTieredCompactionRecoverTask.java | 2 +-
.../inner/sizetiered/SizeTieredCompactionTask.java | 45 ++--
.../db/engine/storagegroup/TsFileResource.java | 5 +
.../engine/compaction/CompactionSchedulerTest.java | 2 +-
.../utils/CompactionFileGeneratorUtils.java | 3 +
7 files changed, 232 insertions(+), 176 deletions(-)
diff --git
a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBNewTsFileCompactionIT.java
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBNewTsFileCompactionIT.java
index 4ccde16..cfe7b40 100644
---
a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBNewTsFileCompactionIT.java
+++
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBNewTsFileCompactionIT.java
@@ -61,7 +61,7 @@ public class IoTDBNewTsFileCompactionIT {
public void setUp() throws Exception {
preMaxNumberOfPointsInPage =
TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
- storageGroupPath = new PartialPath("root.sg1");
+ storageGroupPath = new PartialPath("root.newTsFileCompaction");
TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(1);
originCompactionFileNum =
IoTDBDescriptor.getInstance().getConfig().getMaxCompactionCandidateFileNum();
@@ -76,7 +76,7 @@ public class IoTDBNewTsFileCompactionIT {
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
Statement statement = connection.createStatement()) {
- statement.execute("SET STORAGE GROUP TO root.sg1");
+ statement.execute("SET STORAGE GROUP TO root.newTsFileCompaction");
}
}
@@ -112,22 +112,23 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
statement.execute("FLUSH");
// second file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -166,23 +167,24 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// two chunks
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
// second file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -220,23 +222,24 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
// second file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -277,26 +280,27 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
// second file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -334,24 +338,25 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
statement.execute("FLUSH");
// second file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -391,15 +396,15 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// two chunks
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(2);
// second file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
@@ -407,11 +412,12 @@ public class IoTDBNewTsFileCompactionIT {
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -450,24 +456,25 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
// second file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -509,28 +516,29 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
// second file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(6, 6)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(6, 6)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -568,25 +576,26 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(1);
// second file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -626,25 +635,26 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// two chunks
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
// second file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -653,9 +663,10 @@ public class IoTDBNewTsFileCompactionIT {
assertEquals(retArray.length, cnt);
try (ResultSet resultSet =
- statement.executeQuery("SELECT count(s1) FROM root.sg1.d1 where time
< 4")) {
+ statement.executeQuery(
+ "SELECT count(s1) FROM root.newTsFileCompaction.d1 where time <
4")) {
assertTrue(resultSet.next());
- assertEquals(3L, resultSet.getLong("count(root.sg1.d1.s1)"));
+ assertEquals(3L,
resultSet.getLong("count(root.newTsFileCompaction.d1.s1)"));
}
} catch (StorageEngineException | InterruptedException e) {
e.printStackTrace();
@@ -688,25 +699,26 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(1);
// second file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -747,29 +759,30 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(2);
// first file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(1);
// second file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(6, 6)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(6, 6)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -809,28 +822,29 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(10000);
// first file
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(2);
// second file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -872,29 +886,30 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// two chunks
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(2);
// second file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(6, 6)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(6, 6)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -936,29 +951,30 @@ public class IoTDBNewTsFileCompactionIT {
// first file
// two pages for one chunk
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
statement.execute("FLUSH");
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(2);
// second file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(6, 6)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(6, 6)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
@@ -1001,31 +1017,32 @@ public class IoTDBNewTsFileCompactionIT {
IoTDBDescriptor.getInstance().getConfig().setAvgSeriesPointNumberThreshold(2);
// first file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(1, 1)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(2, 2)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(1, 1)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(2, 2)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(3, 3)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(4, 4)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(3, 3)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(4, 4)");
statement.execute("FLUSH");
// second file
// one chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(5, 5)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(6, 6)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(5, 5)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(6, 6)");
// another chunk with two pages
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(7, 7)");
- statement.execute("INSERT INTO root.sg1.d1(time,s1) values(8, 8)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(7, 7)");
+ statement.execute("INSERT INTO root.newTsFileCompaction.d1(time,s1)
values(8, 8)");
statement.execute("FLUSH");
statement.execute("MERGE");
assertTrue(waitForMergeFinish());
int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM
root.sg1.d1")) {
+ try (ResultSet resultSet =
+ statement.executeQuery("SELECT s1 FROM
root.newTsFileCompaction.d1")) {
cnt = 0;
while (resultSet.next()) {
long time = resultSet.getLong("Time");
- float s1 = resultSet.getFloat("root.sg1.d1.s1");
+ float s1 = resultSet.getFloat("root.newTsFileCompaction.d1.s1");
assertEquals(Long.parseLong(retArray[cnt][0]), time);
assertEquals(Float.parseFloat(retArray[cnt][1]), s1, FLOAT_DELTA);
cnt++;
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionHandler.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionHandler.java
index b5573f8..fa5b380 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionHandler.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionHandler.java
@@ -71,7 +71,11 @@ public class InnerSpaceCompactionExceptionHandler {
if (allSourceFileExist) {
handleSuccess =
handleWhenAllSourceFilesExist(
- fullStorageGroupName, targetTsFile, selectedTsFileResourceList);
+ fullStorageGroupName,
+ targetTsFile,
+ selectedTsFileResourceList,
+ tsFileResourceList,
+ false);
} else {
// some source file does not exists
// it means we start to delete source file
@@ -137,54 +141,74 @@ public class InnerSpaceCompactionExceptionHandler {
public static boolean handleWhenAllSourceFilesExist(
String fullStorageGroupName,
TsFileResource targetTsFile,
- List<TsFileResource> selectedTsFileResourceList) {
- // all source file exists, delete the target file
- LOGGER.info(
- "{} [Compaction][ExceptionHandler] all source files {} exists, delete
target file {}",
- fullStorageGroupName,
- selectedTsFileResourceList,
- targetTsFile);
- TsFileResource tmpTargetTsFile;
- if
(targetTsFile.getTsFilePath().endsWith(IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX))
{
- tmpTargetTsFile = targetTsFile;
- targetTsFile =
- new TsFileResource(
- new File(
- tmpTargetTsFile
- .getTsFilePath()
- .replace(
- IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
- TsFileConstant.TSFILE_SUFFIX)));
- } else {
- tmpTargetTsFile =
- new TsFileResource(
- new File(
- targetTsFile
- .getTsFilePath()
- .replace(
- TsFileConstant.TSFILE_SUFFIX,
- IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
- }
- if (!tmpTargetTsFile.remove()) {
- // failed to remove tmp target tsfile
- // system should not carry out the subsequent compaction in case of data
redundant
- LOGGER.warn(
- "{} [Compaction][ExceptionHandler] failed to remove target file {}",
- fullStorageGroupName,
- tmpTargetTsFile);
- return false;
- }
- if (!targetTsFile.remove()) {
- // failed to remove target tsfile
- // system should not carry out the subsequent compaction in case of data
redundant
- LOGGER.error(
- "{} [Compaction][ExceptionHandler] failed to remove target file {}",
+ List<TsFileResource> selectedTsFileResourceList,
+ TsFileResourceList tsFileResourceList,
+ boolean isRecover) {
+ try {
+ // all source file exists, delete the target file
+ LOGGER.info(
+ "{} [Compaction][ExceptionHandler] all source files {} exists,
delete target file {}",
fullStorageGroupName,
+ selectedTsFileResourceList,
targetTsFile);
- return false;
- }
- // deal with compaction modification
- try {
+ TsFileResource tmpTargetTsFile;
+ if
(targetTsFile.getTsFilePath().endsWith(IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX))
{
+ tmpTargetTsFile = targetTsFile;
+ targetTsFile =
+ new TsFileResource(
+ new File(
+ tmpTargetTsFile
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX)));
+ } else {
+ tmpTargetTsFile =
+ new TsFileResource(
+ new File(
+ targetTsFile
+ .getTsFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX,
+ IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
+ }
+ if (!tmpTargetTsFile.remove()) {
+ // failed to remove tmp target tsfile
+ // system should not carry out the subsequent compaction in case of
data redundant
+ LOGGER.warn(
+ "{} [Compaction][ExceptionHandler] failed to remove target file
{}",
+ fullStorageGroupName,
+ tmpTargetTsFile);
+ return false;
+ }
+ if (!isRecover) {
+ tsFileResourceList.writeLock();
+ try {
+ if (targetTsFile.isFileInList()) {
+ // target tsfile is in the list, remove it
+ tsFileResourceList.remove(targetTsFile);
+ }
+ for (TsFileResource tsFileResource : selectedTsFileResourceList) {
+ // if the source file is not in tsfileResourceList
+ // insert it into the list
+ if (!tsFileResource.isFileInList()) {
+ tsFileResourceList.keepOrderInsert(tsFileResource);
+ }
+ }
+ } finally {
+ tsFileResourceList.writeUnlock();
+ }
+ }
+ if (!targetTsFile.remove()) {
+ // failed to remove target tsfile
+ // system should not carry out the subsequent compaction in case of
data redundant
+ LOGGER.error(
+ "{} [Compaction][ExceptionHandler] failed to remove target file
{}",
+ fullStorageGroupName,
+ targetTsFile);
+ return false;
+ }
+ // deal with compaction modification
InnerSpaceCompactionUtils.appendNewModificationsToOldModsFile(selectedTsFileResourceList);
} catch (Throwable e) {
LOGGER.error(
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
index 957eb3a..169f98f 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
@@ -146,7 +146,7 @@ public class SizeTieredCompactionRecoverTask extends
SizeTieredCompactionTask {
}
handleSuccess =
InnerSpaceCompactionExceptionHandler.handleWhenAllSourceFilesExist(
- fullStorageGroupName, targetResource, sourceResources);
+ fullStorageGroupName, targetResource, sourceResources,
tsFileResourceList, true);
} else {
handleSuccess =
handleWithoutAllSourceFilesExist(sourceFileIdentifiers);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
index 9fa41aa..c1f5130 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
@@ -129,16 +129,39 @@ public class SizeTieredCompactionTask extends
AbstractInnerSpaceCompactionTask {
}
InnerSpaceCompactionUtils.moveTargetFile(targetTsFileResource,
fullStorageGroupName);
+
+ LOGGER.info("{} [SizeTiredCompactionTask] start to rename mods file",
fullStorageGroupName);
+ InnerSpaceCompactionUtils.combineModsInCompaction(
+ selectedTsFileResourceList, targetTsFileResource);
+
LOGGER.info(
- "{} [SizeTiredCompactionTask] compact finish, close the logger",
fullStorageGroupName);
+ "{} [SizeTiredCompactionTask] compact finish, close the logger, edit
the tsFileResourceList",
+ fullStorageGroupName);
sizeTieredCompactionLogger.close();
- LOGGER.info(
- "{} [Compaction] compaction finish, start to delete old files",
fullStorageGroupName);
if (Thread.currentThread().isInterrupted()) {
throw new InterruptedException(
String.format("%s [Compaction] abort", fullStorageGroupName));
}
+
+ // replace the old files with new file, the new is in same position as
the old
+ for (TsFileResource resource : selectedTsFileResourceList) {
+ TsFileResourceManager.getInstance().removeTsFileResource(resource);
+ }
+
TsFileResourceManager.getInstance().registerSealedTsFileResource(targetTsFileResource);
+ tsFileResourceList.writeLock();
+ try {
+ for (TsFileResource resource : selectedTsFileResourceList) {
+ tsFileResourceList.remove(resource);
+ }
+ tsFileResourceList.keepOrderInsert(targetTsFileResource);
+ } finally {
+ tsFileResourceList.writeUnlock();
+ }
+
+ LOGGER.info(
+ "{} [Compaction] compaction finish, start to delete old files",
fullStorageGroupName);
+
LOGGER.info(
"{} [Compaction] Compacted target files, try to get the write lock
of source files",
fullStorageGroupName);
@@ -151,12 +174,6 @@ public class SizeTieredCompactionTask extends
AbstractInnerSpaceCompactionTask {
isHoldingWriteLock[i] = true;
}
- LOGGER.info(
- "{} [SizeTiredCompactionTask] old file deleted, start to rename mods
file",
- fullStorageGroupName);
- InnerSpaceCompactionUtils.combineModsInCompaction(
- selectedTsFileResourceList, targetTsFileResource);
-
if (targetTsFileResource.getTsFile().length()
< TSFileConfig.MAGIC_STRING.getBytes().length * 2L + Byte.BYTES) {
// the file size is smaller than magic string and version number
@@ -172,16 +189,6 @@ public class SizeTieredCompactionTask extends
AbstractInnerSpaceCompactionTask {
InnerSpaceCompactionUtils.deleteModificationForSourceFile(
selectedTsFileResourceList, fullStorageGroupName);
- // replace the old files with new file, the new is in same position as
the old
- for (TsFileResource resource : selectedTsFileResourceList) {
- TsFileResourceManager.getInstance().removeTsFileResource(resource);
- }
- tsFileResourceList.insertBefore(selectedTsFileResourceList.get(0),
targetTsFileResource);
-
TsFileResourceManager.getInstance().registerSealedTsFileResource(targetTsFileResource);
- for (TsFileResource resource : selectedTsFileResourceList) {
- tsFileResourceList.remove(resource);
- }
-
long costTime = System.currentTimeMillis() - startTime;
LOGGER.info(
"{} [SizeTiredCompactionTask] all compaction task finish, target
file is {},"
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 57f0a1c..0d4c1b0 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -978,4 +978,9 @@ public class TsFileResource {
pathToReadOnlyMemChunkMap.get(path),
pathToChunkMetadataListMap.get(path)));
}
}
+
+ /** @return is this tsfile resource in a TsFileResourceList */
+ public boolean isFileInList() {
+ return prev != null || next != null;
+ }
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionSchedulerTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionSchedulerTest.java
index 878ac7d..d03f2e7 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionSchedulerTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionSchedulerTest.java
@@ -59,7 +59,7 @@ import static org.junit.Assert.fail;
public class CompactionSchedulerTest {
private static final Logger logger =
LoggerFactory.getLogger(CompactionSchedulerTest.class);
- static final String COMPACTION_TEST_SG = "root.compactionTest";
+ static final String COMPACTION_TEST_SG = "root.compactionSchedulerTest";
private static final boolean oldEnableInnerSeqCompaction =
IoTDBDescriptor.getInstance().getConfig().isEnableSeqSpaceCompaction();
private static final boolean oldEnableInnerUnseqCompaction =
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionFileGeneratorUtils.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionFileGeneratorUtils.java
index 6345279..dc9d345 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionFileGeneratorUtils.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionFileGeneratorUtils.java
@@ -174,6 +174,9 @@ public class CompactionFileGeneratorUtils {
TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(Integer.MAX_VALUE);
+ if (!newTsFileResource.getTsFile().getParentFile().exists()) {
+ newTsFileResource.getTsFile().getParentFile().mkdirs();
+ }
RestorableTsFileIOWriter writer = new
RestorableTsFileIOWriter(newTsFileResource.getTsFile());
Map<String, List<String>> deviceMeasurementMap = new HashMap<>();
for (String fullPath : fullPaths) {
