You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/11/30 08:31:26 UTC
[iotdb] branch master updated: Fix compaction add list bug when
last level exceeds max file num (#2141)
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 7ebdf9a Fix compaction add list bug when last level exceeds max file num (#2141)
7ebdf9a is described below
commit 7ebdf9ab2571fe6bd7d1192de84d7f2600449d32
Author: zhanglingzhe0820 <44...@qq.com>
AuthorDate: Mon Nov 30 16:31:06 2020 +0800
Fix compaction add list bug when last level exceeds max file num (#2141)
---
.../db/engine/compaction/TsFileManagement.java | 21 +++++
.../level/LevelCompactionTsFileManagement.java | 7 +-
.../no/NoCompactionTsFileManagement.java | 22 +----
.../iotdb/db/integration/IoTDBCompactionIT.java | 99 ++++++++++++++++++++++
.../iotdb/db/integration/IoTDBRestartIT.java | 61 ++++++++++++-
5 files changed, 185 insertions(+), 25 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
index 70e8526..d10b4c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
@@ -19,7 +19,9 @@
package org.apache.iotdb.db.engine.compaction;
+import static org.apache.iotdb.db.conf.IoTDBConstant.FILE_NAME_SEPARATOR;
import static org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.MERGING_MODIFICATION_FILE_NAME;
+import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
import java.io.File;
import java.io.IOException;
@@ -380,4 +382,23 @@ public abstract class TsFileManagement {
logger.info("{} a merge task ends", storageGroupName);
}
+ // ({systemTime}-{versionNum}-{mergeNum}.tsfile)
+ public static int compareFileName(File o1, File o2) {
+ String[] items1 = o1.getName().replace(TSFILE_SUFFIX, "")
+ .split(FILE_NAME_SEPARATOR);
+ String[] items2 = o2.getName().replace(TSFILE_SUFFIX, "")
+ .split(FILE_NAME_SEPARATOR);
+ long ver1 = Long.parseLong(items1[0]);
+ long ver2 = Long.parseLong(items2[0]);
+ int cmp = Long.compare(ver1, ver2);
+ if (cmp == 0) {
+ int cmpVersion = Long.compare(Long.parseLong(items1[1]), Long.parseLong(items2[1]));
+ if (cmpVersion == 0) {
+ return Long.compare(Long.parseLong(items1[2]), Long.parseLong(items2[2]));
+ }
+ return cmpVersion;
+ } else {
+ return cmp;
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
index 515ef6b..2881cb6 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
@@ -92,16 +92,17 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
}
}
- private void deleteLevelFilesInList(long timePartitionId, Collection<TsFileResource> mergeTsFiles) {
+ private void deleteLevelFilesInList(long timePartitionId,
+ Collection<TsFileResource> mergeTsFiles) {
logger.debug("{} [compaction] merge starts to delete file list", storageGroupName);
- for (int i = 0; i < seqLevelNum; i++) {
+ for (int i = 0; i < seqLevelNum - 1; i++) {
if (sequenceTsFileResources.containsKey(timePartitionId)) {
if (sequenceTsFileResources.get(timePartitionId).size() > i) {
sequenceTsFileResources.get(timePartitionId).get(i).removeAll(mergeTsFiles);
}
}
}
- for (int i = 0; i < unseqLevelNum; i++) {
+ for (int i = 0; i < unseqLevelNum - 1; i++) {
if (unSequenceTsFileResources.containsKey(timePartitionId)) {
if (unSequenceTsFileResources.get(timePartitionId).size() > i) {
unSequenceTsFileResources.get(timePartitionId).get(i).removeAll(mergeTsFiles);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
index 935a5c0..dec92e0 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
@@ -19,16 +19,12 @@
package org.apache.iotdb.db.engine.compaction.no;
-import static org.apache.iotdb.db.conf.IoTDBConstant.FILE_NAME_SEPARATOR;
-import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
-
-import java.io.File;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.compaction.TsFileManagement;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,22 +46,6 @@ public class NoCompactionTsFileManagement extends TsFileManagement {
super(storageGroupName, storageGroupDir);
}
- // ({systemTime}-{versionNum}-{mergeNum}.tsfile)
- public static int compareFileName(File o1, File o2) {
- String[] items1 = o1.getName().replace(TSFILE_SUFFIX, "")
- .split(FILE_NAME_SEPARATOR);
- String[] items2 = o2.getName().replace(TSFILE_SUFFIX, "")
- .split(FILE_NAME_SEPARATOR);
- long ver1 = Long.parseLong(items1[0]);
- long ver2 = Long.parseLong(items2[0]);
- int cmp = Long.compare(ver1, ver2);
- if (cmp == 0) {
- return Long.compare(Long.parseLong(items1[1]), Long.parseLong(items2[1]));
- } else {
- return cmp;
- }
- }
-
@Override
public List<TsFileResource> getStableTsFileList(boolean sequence) {
return getTsFileList(sequence);
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCompactionIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCompactionIT.java
new file mode 100644
index 0000000..d3cc788
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCompactionIT.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.integration;
+
+import static org.junit.Assert.assertEquals;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class IoTDBCompactionIT {
+
+ private int prevSeqLevelFileNum;
+ private int prevSeqLevelNum;
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ prevSeqLevelFileNum = IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ prevSeqLevelNum = IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ @Test
+ public void test() throws SQLException {
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + " WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 0; i < 32; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(32, cnt);
+ }
+ }
+
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java
index 5b42f1e..c40f9b2 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java
@@ -95,7 +95,6 @@ public class IoTDBRestartIT {
EnvironmentUtils.cleanEnv();
}
-
@Test
public void testRestartDelete()
throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
@@ -350,4 +349,64 @@ public class IoTDBRestartIT {
EnvironmentUtils.cleanEnv();
}
+
+ @Test
+ public void testRestartCompaction()
+ throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
+ "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(2,1.0)");
+ statement.execute("flush");
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(3,1.0)");
+ statement.execute("flush");
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(4,1.0)");
+ statement.execute("flush");
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(5,1.0)");
+ statement.execute("flush");
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(6,1.0)");
+ statement.execute("flush");
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(1,1.0)");
+ statement.execute("flush");
+ statement.execute("insert into root.turbine.d1(timestamp,s1) values(7,1.0)");
+ statement.execute("flush");
+ }
+
+ try {
+ EnvironmentUtils.restartDaemon();
+ } catch (Exception e) {
+ Assert.fail();
+ }
+
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
+ "root");
+ Statement statement = connection.createStatement()) {
+ boolean hasResultSet = statement.execute("SELECT s1 FROM root.turbine.d1");
+ assertTrue(hasResultSet);
+ String[] exp = new String[]{
+ "1,1.0",
+ "2,1.0",
+ "3,1.0",
+ "4,1.0",
+ "5,1.0",
+ "6,1.0",
+ "7,1.0"
+ };
+ int cnt = 0;
+ try (ResultSet resultSet = statement.getResultSet()) {
+ while (resultSet.next()) {
+ String result = resultSet.getString(TIMESTAMP_STR) + "," + resultSet.getString(2);
+ assertEquals(exp[cnt], result);
+ cnt++;
+ }
+ }
+ }
+
+ EnvironmentUtils.cleanEnv();
+ }
}