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/09/09 13:33:04 UTC
[incubator-iotdb] branch master updated: [IOTDB-875]fix flush
storage group error when no data in storage group (#1700)
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/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 2c4fae5 [IOTDB-875]fix flush storage group error when no data in storage group (#1700)
2c4fae5 is described below
commit 2c4fae58b0f6e30e06fbc80b64be5444dfcb2438
Author: HouliangQi <ne...@163.com>
AuthorDate: Wed Sep 9 21:32:52 2020 +0800
[IOTDB-875]fix flush storage group error when no data in storage group (#1700)
---
.../org/apache/iotdb/db/engine/StorageEngine.java | 70 ++++++++++++++--------
.../org/apache/iotdb/db/metadata/MManager.java | 2 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 61 +++++++++++++++----
.../db/integration/IoTDBFlushQueryMergeIT.java | 61 +++++++++++++++++--
4 files changed, 152 insertions(+), 42 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index a1ba488..88cd7c7 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -18,6 +18,23 @@
*/
package org.apache.iotdb.db.engine;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.ConcurrentModificationException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -31,7 +48,13 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.exception.*;
+import org.apache.iotdb.db.exception.BatchInsertionException;
+import org.apache.iotdb.db.exception.LoadFileException;
+import org.apache.iotdb.db.exception.ShutdownException;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.StorageGroupProcessorException;
+import org.apache.iotdb.db.exception.TsFileProcessorException;
+import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
@@ -55,13 +78,6 @@ import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.File;
-import java.io.IOException;
-import java.util.*;
-import java.util.Map.Entry;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
-
public class StorageEngine implements IService {
private final Logger logger;
@@ -117,8 +133,7 @@ public class StorageEngine implements IService {
private static long timePartitionInterval = -1;
/**
- * whether enable data partition
- * if disabled, all data belongs to partition 0
+ * whether enable data partition if disabled, all data belongs to partition 0
*/
@ServerConfigConsistent
private static boolean enablePartition =
@@ -149,7 +164,7 @@ public class StorageEngine implements IService {
public void recover() {
recoverAllSgThreadPool = IoTDBThreadPoolFactory
- .newSingleThreadExecutor("Begin-Recovery-Pool");
+ .newSingleThreadExecutor("Begin-Recovery-Pool");
recoverAllSgThreadPool.submit(this::recoverAllSgs);
}
@@ -163,13 +178,14 @@ public class StorageEngine implements IService {
futures.add(recoveryThreadPool.submit((Callable<Void>) () -> {
try {
StorageGroupProcessor processor = new StorageGroupProcessor(systemDir,
- storageGroup.getFullPath(), fileFlushPolicy);
+ storageGroup.getFullPath(), fileFlushPolicy);
processor.setDataTTL(storageGroup.getDataTTL());
processorMap.put(storageGroup.getPartialPath(), processor);
logger.info("Storage Group Processor {} is recovered successfully",
- storageGroup.getFullPath());
+ storageGroup.getFullPath());
} catch (Exception e) {
- logger.error("meet error when recovering storage group: {}", storageGroup.getFullPath(), e);
+ logger
+ .error("meet error when recovering storage group: {}", storageGroup.getFullPath(), e);
}
return null;
}));
@@ -291,15 +307,17 @@ public class StorageEngine implements IService {
processor = processorMap.get(storageGroupPath);
if (processor == null) {
logger.info("construct a processor instance, the storage group is {}, Thread is {}",
- storageGroupPath, Thread.currentThread().getId());
- processor = new StorageGroupProcessor(systemDir, storageGroupPath.getFullPath(), fileFlushPolicy);
+ storageGroupPath, Thread.currentThread().getId());
+ processor = new StorageGroupProcessor(systemDir, storageGroupPath.getFullPath(),
+ fileFlushPolicy);
processor.setDataTTL(storageGroupMNode.getDataTTL());
processorMap.put(storageGroupPath, processor);
}
}
} else {
// not finished recover, refuse the request
- throw new StorageEngineException("the sg " + storageGroupPath + " may not ready now, please wait and retry later",
+ throw new StorageEngineException(
+ "the sg " + storageGroupPath + " may not ready now, please wait and retry later",
TSStatusCode.STORAGE_GROUP_NOT_READY.getStatusCode());
}
}
@@ -369,8 +387,7 @@ public class StorageEngine implements IService {
}
}
- public void asyncCloseProcessor(PartialPath storageGroupPath, boolean isSeq)
- throws StorageGroupNotSetException {
+ public void asyncCloseProcessor(PartialPath storageGroupPath, boolean isSeq) {
StorageGroupProcessor processor = processorMap.get(storageGroupPath);
if (processor != null) {
logger.info("async closing sg processor is called for closing {}, seq = {}", storageGroupPath,
@@ -393,8 +410,6 @@ public class StorageEngine implements IService {
} finally {
processor.writeUnlock();
}
- } else {
- throw new StorageGroupNotSetException(storageGroupPath.getFullPath());
}
}
@@ -575,17 +590,20 @@ public class StorageEngine implements IService {
public boolean deleteTsfileForSync(File deletedTsfile)
throws StorageEngineException, IllegalPathException {
- return getProcessor(new PartialPath(deletedTsfile.getParentFile().getName())).deleteTsfile(deletedTsfile);
+ return getProcessor(new PartialPath(deletedTsfile.getParentFile().getName()))
+ .deleteTsfile(deletedTsfile);
}
public boolean deleteTsfile(File deletedTsfile)
throws StorageEngineException, IllegalPathException {
- return getProcessor(new PartialPath(getSgByEngineFile(deletedTsfile))).deleteTsfile(deletedTsfile);
+ return getProcessor(new PartialPath(getSgByEngineFile(deletedTsfile)))
+ .deleteTsfile(deletedTsfile);
}
public boolean moveTsfile(File tsfileToBeMoved, File targetDir)
throws StorageEngineException, IllegalPathException {
- return getProcessor(new PartialPath(getSgByEngineFile(tsfileToBeMoved))).moveTsfile(tsfileToBeMoved, targetDir);
+ return getProcessor(new PartialPath(getSgByEngineFile(tsfileToBeMoved)))
+ .moveTsfile(tsfileToBeMoved, targetDir);
}
/**
@@ -648,11 +666,13 @@ public class StorageEngine implements IService {
/**
* Set the version of given partition to newMaxVersion if it is larger than the current version.
+ *
* @param storageGroup
* @param partitionId
* @param newMaxVersion
*/
- public void setPartitionVersionToMax(PartialPath storageGroup, long partitionId, long newMaxVersion)
+ public void setPartitionVersionToMax(PartialPath storageGroup, long partitionId,
+ long newMaxVersion)
throws StorageEngineException {
getProcessor(storageGroup).setPartitionFileVersionToMax(partitionId, newMaxVersion);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index 67c0681..f9eb96e 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -653,7 +653,7 @@ public class MManager {
* @param path Format: root.node.(node)*
* @apiNote :for cluster
*/
- boolean isStorageGroup(PartialPath path) {
+ public boolean isStorageGroup(PartialPath path) {
lock.readLock().lock();
try {
return mtree.isStorageGroup(path);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index 8050ddf..2086c9e 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -257,7 +257,8 @@ public class PlanExecutor implements IPlanExecutor {
DeletePartitionPlan p = (DeletePartitionPlan) plan;
TimePartitionFilter filter =
(storageGroupName, partitionId) ->
- storageGroupName.equals(((DeletePartitionPlan) plan).getStorageGroupName().getFullPath())
+ storageGroupName
+ .equals(((DeletePartitionPlan) plan).getStorageGroupName().getFullPath())
&& p.getPartitionId().contains(partitionId);
StorageEngine.getInstance()
.removePartitions(((DeletePartitionPlan) plan).getStorageGroupName(), filter);
@@ -309,6 +310,16 @@ public class PlanExecutor implements IPlanExecutor {
}
}
}
+
+ if (plan.getPaths() != null) {
+ List<PartialPath> noExistSg = checkStorageGroupExist(plan.getPaths());
+ if (!noExistSg.isEmpty()) {
+ StringBuilder sb = new StringBuilder();
+ noExistSg.forEach((storageGroup) -> sb.append(storageGroup.getFullPath()).append(","));
+ throw new StorageGroupNotSetException(
+ sb.subSequence(0, sb.length() - 1).toString());
+ }
+ }
}
protected QueryDataSet processDataQuery(QueryPlan queryPlan, QueryContext context)
@@ -400,7 +411,8 @@ public class PlanExecutor implements IPlanExecutor {
List<PartialPath> nodes = getNodesList(countPlan.getPath(), countPlan.getLevel());
ListDataSet listDataSet =
new ListDataSet(
- Arrays.asList(new PartialPath(COLUMN_COLUMN, false), new PartialPath(COLUMN_COUNT, false)),
+ Arrays.asList(new PartialPath(COLUMN_COLUMN, false),
+ new PartialPath(COLUMN_COUNT, false)),
Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
for (PartialPath columnPath : nodes) {
RowRecord record = new RowRecord(0);
@@ -428,7 +440,8 @@ public class PlanExecutor implements IPlanExecutor {
return IoTDB.metaManager.getAllTimeseriesPath(path);
}
- protected List<PartialPath> getNodesList(PartialPath schemaPattern, int level) throws MetadataException {
+ protected List<PartialPath> getNodesList(PartialPath schemaPattern, int level)
+ throws MetadataException {
return IoTDB.metaManager.getNodesList(schemaPattern, level);
}
@@ -527,7 +540,8 @@ public class PlanExecutor implements IPlanExecutor {
private QueryDataSet processShowTTLQuery(ShowTTLPlan showTTLPlan) {
ListDataSet listDataSet =
new ListDataSet(
- Arrays.asList(new PartialPath(COLUMN_STORAGE_GROUP, false), new PartialPath(COLUMN_TTL, false)),
+ Arrays.asList(new PartialPath(COLUMN_STORAGE_GROUP, false),
+ new PartialPath(COLUMN_TTL, false)),
Arrays.asList(TSDataType.TEXT, TSDataType.INT64));
List<PartialPath> selectedSgs = showTTLPlan.getStorageGroups();
@@ -572,7 +586,8 @@ public class PlanExecutor implements IPlanExecutor {
private QueryDataSet processShowDynamicParameterQuery() {
ListDataSet listDataSet =
new ListDataSet(
- Arrays.asList(new PartialPath(COLUMN_PARAMETER, false), new PartialPath(COLUMN_VALUE, false)),
+ Arrays.asList(new PartialPath(COLUMN_PARAMETER, false),
+ new PartialPath(COLUMN_VALUE, false)),
Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
int timestamp = 0;
@@ -617,7 +632,8 @@ public class PlanExecutor implements IPlanExecutor {
private QueryDataSet processShowFlushTaskInfo() {
ListDataSet listDataSet =
new ListDataSet(
- Arrays.asList(new PartialPath(COLUMN_ITEM, false), new PartialPath(COLUMN_VALUE, false)),
+ Arrays
+ .asList(new PartialPath(COLUMN_ITEM, false), new PartialPath(COLUMN_VALUE, false)),
Arrays.asList(TSDataType.TEXT, TSDataType.TEXT));
int timestamp = 0;
@@ -756,12 +772,16 @@ public class PlanExecutor implements IPlanExecutor {
String device = chunkGroupMetadata.getDevice();
MNode node = null;
try {
- node = mManager.getDeviceNodeWithAutoCreateAndReadLock(new PartialPath(device), true, sgLevel);
+ node = mManager
+ .getDeviceNodeWithAutoCreateAndReadLock(new PartialPath(device), true, sgLevel);
for (ChunkMetadata chunkMetadata : chunkGroupMetadata.getChunkMetadataList()) {
- PartialPath series = new PartialPath(chunkGroupMetadata.getDevice() + TsFileConstant.PATH_SEPARATOR + chunkMetadata.getMeasurementUid());
+ PartialPath series = new PartialPath(
+ chunkGroupMetadata.getDevice() + TsFileConstant.PATH_SEPARATOR + chunkMetadata
+ .getMeasurementUid());
if (!registeredSeries.contains(series)) {
registeredSeries.add(series);
- MeasurementSchema schema = knownSchemas.get(new Path(series.getDevice(), series.getMeasurement()));
+ MeasurementSchema schema = knownSchemas
+ .get(new Path(series.getDevice(), series.getMeasurement()));
if (schema == null) {
throw new MetadataException(
String.format(
@@ -1181,7 +1201,8 @@ public class PlanExecutor implements IPlanExecutor {
}
}
- private ListDataSet executeListRolePrivileges(String roleName, PartialPath path) throws AuthException {
+ private ListDataSet executeListRolePrivileges(String roleName, PartialPath path)
+ throws AuthException {
Role role = authorizer.getRole(roleName);
if (role != null) {
List<PartialPath> headerList = new ArrayList<>();
@@ -1205,7 +1226,8 @@ public class PlanExecutor implements IPlanExecutor {
}
}
- private ListDataSet executeListUserPrivileges(String userName, PartialPath path) throws AuthException {
+ private ListDataSet executeListUserPrivileges(String userName, PartialPath path)
+ throws AuthException {
User user = authorizer.getUser(userName);
if (user == null) {
throw new AuthException("No such user : " + userName);
@@ -1298,4 +1320,21 @@ public class PlanExecutor implements IPlanExecutor {
record.addField(status.isDone(), TSDataType.BOOLEAN);
return record;
}
+
+ /**
+ * @param storageGroups the storage groups to check
+ * @return List<PartialPath> the storage groups that not exist
+ */
+ List<PartialPath> checkStorageGroupExist(List<PartialPath> storageGroups) {
+ List<PartialPath> noExistSg = new ArrayList<>();
+ if (storageGroups == null) {
+ return noExistSg;
+ }
+ for (PartialPath storageGroup : storageGroups) {
+ if (!MManager.getInstance().isStorageGroup(storageGroup)) {
+ noExistSg.add(storageGroup);
+ }
+ }
+ return noExistSg;
+ }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeIT.java
index 79ab3cd..00e57e7 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBFlushQueryMergeIT.java
@@ -24,18 +24,23 @@ import static org.junit.Assert.fail;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
import java.util.Locale;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.metadata.StorageGroupNotSetException;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class IoTDBFlushQueryMergeIT {
+ private static final Logger logger = LoggerFactory.getLogger(IoTDBFlushQueryMergeIT.class);
private static String[] sqls = new String[]{
"SET STORAGE GROUP TO root.vehicle.d0",
"CREATE TIMESERIES root.vehicle.d0.s0 WITH DATATYPE=INT32, ENCODING=RLE",
@@ -79,7 +84,7 @@ public class IoTDBFlushQueryMergeIT {
statement.execute(sql);
}
} catch (Exception e) {
- e.printStackTrace();
+ logger.error("insertData failed", e);
}
}
@@ -101,7 +106,7 @@ public class IoTDBFlushQueryMergeIT {
}
statement.execute("merge");
} catch (Exception e) {
- e.printStackTrace();
+ logger.error("selectAllSQLTest failed", e);
fail(e.getMessage());
}
}
@@ -142,7 +147,8 @@ public class IoTDBFlushQueryMergeIT {
statement.execute("FLUSH root.group2,root.group3 FALSE");
int i = 0;
- try (ResultSet resultSet = statement.executeQuery("SELECT * FROM root.group1,root.group2,root" + ".group3")) {
+ try (ResultSet resultSet = statement
+ .executeQuery("SELECT * FROM root.group1,root.group2,root" + ".group3")) {
while (resultSet.next()) {
i++;
}
@@ -150,7 +156,52 @@ public class IoTDBFlushQueryMergeIT {
assertEquals(30, i);
} catch (Exception e) {
- e.printStackTrace();
+ logger.error("testFlushGivenGroup failed", e);
+ fail(e.getMessage());
+ }
+ }
+
+ // bug fix test, https://issues.apache.org/jira/browse/IOTDB-875
+ @Test
+ public void testFlushGivenGroupNoData() throws ClassNotFoundException {
+ 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("SET STORAGE GROUP TO root.nodatagroup1");
+ statement.execute("SET STORAGE GROUP TO root.nodatagroup2");
+ statement.execute("SET STORAGE GROUP TO root.nodatagroup3");
+ statement.execute("FLUSH root.nodatagroup1");
+ statement.execute("FLUSH root.nodatagroup2");
+ statement.execute("FLUSH root.nodatagroup3");
+ statement.execute("FLUSH root.nodatagroup1, root.nodatagroup2");
+ } catch (Exception e) {
+ logger.error("testFlushGivenGroupNoData failed", e);
+ fail(e.getMessage());
+ }
+ }
+
+ // bug fix test, https://issues.apache.org/jira/browse/IOTDB-875
+ @Test
+ public void testFlushNotExistGroupNoData() throws ClassNotFoundException {
+ 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("SET STORAGE GROUP TO root.noexist.nodatagroup1");
+ try {
+ statement
+ .execute("FLUSH root.noexist.nodatagroup1,root.notExistGroup1,root.notExistGroup2");
+ } catch (SQLException sqe) {
+ StorageGroupNotSetException tmpsgnse = new StorageGroupNotSetException(
+ "root.notExistGroup1,root.notExistGroup2");
+ SQLException sqlException = new SQLException(
+ TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode() + ": " +
+ tmpsgnse.getMessage());
+ assertEquals(sqlException.getMessage(), sqe.getMessage());
+ }
+ } catch (Exception e) {
+ logger.error("testFlushNotExistGroupNoData failed", e);
fail(e.getMessage());
}
}