This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch rel/0.13
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.13 by this push:
new ebc0a9e5f5 [To rel/0.13][IOTDB-3392][IOTDB-3515] File doesn't exist
when unload/move tsfile when virtual_storage_group_num > 1 and update error
message (#6306)
ebc0a9e5f5 is described below
commit ebc0a9e5f56401c2551c731c2e4cc0c232699d95
Author: Chen YZ <[email protected]>
AuthorDate: Fri Jun 17 14:52:59 2022 +0800
[To rel/0.13][IOTDB-3392][IOTDB-3515] File doesn't exist when unload/move
tsfile when virtual_storage_group_num > 1 and update error message (#6306)
---
.../db/integration/IoTDBLoadExternalTsfileIT.java | 8 +-
.../IoTDBLoadExternalTsfileWithVirtualSGIT.java | 184 +++++++++++++++++++++
.../org/apache/iotdb/db/engine/StorageEngine.java | 88 +++++++++-
.../virtualSg/StorageGroupManager.java | 36 ++++
4 files changed, 309 insertions(+), 7 deletions(-)
diff --git
a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileIT.java
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileIT.java
index 3f6bb5016c..d0abb5a843 100644
---
a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileIT.java
+++
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileIT.java
@@ -58,7 +58,7 @@ public class IoTDBLoadExternalTsfileIT {
private static final IoTDBConfig config =
IoTDBDescriptor.getInstance().getConfig();
- private static String[] insertSequenceSqls =
+ protected static String[] insertSequenceSqls =
new String[] {
"SET STORAGE GROUP TO root.vehicle",
"SET STORAGE GROUP TO root.test",
@@ -128,8 +128,8 @@ public class IoTDBLoadExternalTsfileIT {
private static final String TEST_D0_S1_STR = "root.test.d0.s1";
private static final String TEST_D1_STR = "root.test.d1.g0.s0";
- private int prevVirtualPartitionNum;
- private int prevCompactionThread;
+ protected int prevVirtualPartitionNum;
+ protected int prevCompactionThread;
private static String[] deleteSqls =
new String[] {"DELETE STORAGE GROUP root.vehicle", "DELETE STORAGE GROUP
root.test"};
@@ -1012,7 +1012,7 @@ public class IoTDBLoadExternalTsfileIT {
}
}
- private void prepareData(String[] sqls) {
+ protected void prepareData(String[] sqls) {
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
diff --git
a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileWithVirtualSGIT.java
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileWithVirtualSGIT.java
new file mode 100644
index 0000000000..4d99985d11
--- /dev/null
+++
b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBLoadExternalTsfileWithVirtualSGIT.java
@@ -0,0 +1,184 @@
+/*
+ * 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 org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.apache.iotdb.tsfile.utils.FilePathUtils;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class IoTDBLoadExternalTsfileWithVirtualSGIT extends
IoTDBLoadExternalTsfileIT {
+ @Before
+ public void setUp() throws Exception {
+ prevVirtualPartitionNum =
IoTDBDescriptor.getInstance().getConfig().getVirtualStorageGroupNum();
+ IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(2);
+ prevCompactionThread =
+
IoTDBDescriptor.getInstance().getConfig().getConcurrentCompactionThread();
+ EnvironmentUtils.envSetUp();
+ StorageEngine.getInstance().reset();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ prepareData(insertSequenceSqls);
+ }
+
+ @Test
+ public void unloadTsfileWithVSGTest() throws SQLException {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/",
"root", "root");
+ Statement statement = connection.createStatement()) {
+
+ // move root.vehicle
+ File vehicleDir =
+ new File(
+ IoTDBDescriptor.getInstance().getConfig().getDataDirs()[0],
+ IoTDBConstant.SEQUENCE_FLODER_NAME + File.separator +
"root.vehicle");
+ List<File> vehicleFiles = getTsFilePaths(vehicleDir);
+ File tmpDir =
+ new File(
+ IoTDBDescriptor.getInstance().getConfig().getDataDirs()[0],
+ "tmp" + File.separator + new PartialPath("root.vehicle"));
+ if (!tmpDir.exists()) {
+ tmpDir.mkdirs();
+ }
+ for (File tsFile : vehicleFiles) {
+ statement.execute(String.format("unload \"%s\" \"%s\"",
tsFile.getAbsolutePath(), tmpDir));
+ }
+ assertEquals(0, getTsFilePaths(vehicleDir).size());
+ assertNotNull(tmpDir.listFiles());
+ assertEquals(2, tmpDir.listFiles().length >> 1);
+ //
+ // // move root.test
+ File testDir =
+ new File(
+ IoTDBDescriptor.getInstance().getConfig().getDataDirs()[0],
+ IoTDBConstant.SEQUENCE_FLODER_NAME + File.separator +
"root.test");
+ List<File> testFiles = getTsFilePaths(testDir);
+ tmpDir =
+ new File(
+ IoTDBDescriptor.getInstance().getConfig().getDataDirs()[0],
+ "tmp" + File.separator + new PartialPath("root.test"));
+ if (!tmpDir.exists()) {
+ tmpDir.mkdirs();
+ }
+ for (File tsFile : testFiles) {
+ statement.execute(String.format("unload \"%s\" \"%s\"",
tsFile.getAbsolutePath(), tmpDir));
+ }
+ assertEquals(0, getTsFilePaths(testDir).size());
+ assertNotNull(tmpDir.listFiles());
+ assertEquals(2, tmpDir.listFiles().length >> 1);
+ } catch (IllegalPathException e) {
+ Assert.fail();
+ }
+ }
+
+ @Test
+ public void removeTsfileWithVSGTest() throws SQLException {
+ try (Connection connection =
+ DriverManager.getConnection("jdbc:iotdb://127.0.0.1:6667/",
"root", "root");
+ Statement statement = connection.createStatement()) {
+
+ // remove root.vehicle
+ File vehicleDir =
+ new File(
+ IoTDBDescriptor.getInstance().getConfig().getDataDirs()[0],
+ IoTDBConstant.SEQUENCE_FLODER_NAME + File.separator +
"root.vehicle");
+ List<File> vehicleFiles = getTsFilePaths(vehicleDir);
+ for (File tsFile : vehicleFiles) {
+ statement.execute(String.format("remove \"%s\"",
tsFile.getAbsolutePath()));
+ }
+ assertEquals(0, getTsFilePaths(vehicleDir).size());
+ // remove root.test
+ File testDir =
+ new File(
+ IoTDBDescriptor.getInstance().getConfig().getDataDirs()[0],
+ IoTDBConstant.SEQUENCE_FLODER_NAME + File.separator +
"root.test");
+ List<File> testFiles = getTsFilePaths(testDir);
+ for (File tsFile : testFiles) {
+ statement.execute(String.format("remove \"%s\"",
tsFile.getAbsolutePath()));
+ }
+ assertEquals(0, getTsFilePaths(testDir).size());
+ }
+ }
+
+ /**
+ * scan parentDir and return all TsFile sorted by load sequence
+ *
+ * @param parentDir folder to scan
+ */
+ public static List<File> getTsFilePaths(File parentDir) {
+ List<File> res = new ArrayList<>();
+ if (!parentDir.exists()) {
+ Assert.fail();
+ return res;
+ }
+ scanDir(res, parentDir);
+ res.sort(
+ (f1, f2) -> {
+ int diffSg =
+ f1.getParentFile()
+ .getParentFile()
+ .getParentFile()
+ .getName()
+
.compareTo(f2.getParentFile().getParentFile().getParentFile().getName());
+ if (diffSg != 0) {
+ return diffSg;
+ } else {
+ return (int)
+ (FilePathUtils.splitAndGetTsFileVersion(f1.getName())
+ - FilePathUtils.splitAndGetTsFileVersion(f2.getName()));
+ }
+ });
+ return res;
+ }
+
+ private static void scanDir(List<File> tsFiles, File parentDir) {
+ if (!parentDir.exists()) {
+ Assert.fail();
+ return;
+ }
+ File fa[] = parentDir.listFiles();
+ for (int i = 0; i < fa.length; i++) {
+ File fs = fa[i];
+ if (fs.isDirectory()) {
+ scanDir(tsFiles, fs);
+ } else if (fs.getName().endsWith(".resource")) {
+ // only add tsfile that has been flushed
+ tsFiles.add(new File(fs.getAbsolutePath().substring(0,
fs.getAbsolutePath().length() - 9)));
+ }
+ }
+ }
+}
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 413518f963..a1c6a056d9 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
@@ -467,6 +467,24 @@ public class StorageEngine implements IService {
}
}
+ /**
+ * This method is for sync, delete tsfile or sth like them, just get storage
group directly by
+ * virtualStorageGroupId
+ *
+ * @param path storage group path
+ * @param virtualStorageGroupId virtual storage group partition id
+ * @return storage group processor
+ */
+ public VirtualStorageGroupProcessor getProcessorDirectly(
+ PartialPath path, int virtualStorageGroupId) throws
StorageEngineException {
+ try {
+ IStorageGroupMNode storageGroupMNode =
IoTDB.metaManager.getStorageGroupNodeByPath(path);
+ return getStorageGroupProcessorById(virtualStorageGroupId,
storageGroupMNode);
+ } catch (StorageGroupProcessorException | MetadataException e) {
+ throw new StorageEngineException(e);
+ }
+ }
+
/**
* This method is for insert and query or sth like them, this may get a
virtual storage group
*
@@ -528,6 +546,32 @@ public class StorageEngine implements IService {
return storageGroupManager.getProcessor(devicePath, storageGroupMNode);
}
+ /**
+ * get storage group processor by virtualStorageGroupId
+ *
+ * @param virtualStorageGroupId virtual storage group partition id
+ * @param storageGroupMNode mnode of the storage group, we need synchronize
this to avoid
+ * modification in mtree
+ * @return found or new storage group processor
+ */
+ @SuppressWarnings("java:S2445")
+ // actually storageGroupMNode is a unique object on the mtree, synchronize
it is reasonable
+ private VirtualStorageGroupProcessor getStorageGroupProcessorById(
+ int virtualStorageGroupId, IStorageGroupMNode storageGroupMNode)
+ throws StorageGroupProcessorException, StorageEngineException {
+ StorageGroupManager storageGroupManager =
processorMap.get(storageGroupMNode.getPartialPath());
+ if (storageGroupManager == null) {
+ synchronized (this) {
+ storageGroupManager =
processorMap.get(storageGroupMNode.getPartialPath());
+ if (storageGroupManager == null) {
+ storageGroupManager = new StorageGroupManager();
+ processorMap.put(storageGroupMNode.getPartialPath(),
storageGroupManager);
+ }
+ }
+ }
+ return storageGroupManager.getProcessor(virtualStorageGroupId,
storageGroupMNode);
+ }
+
/**
* build a new storage group processor
*
@@ -833,12 +877,14 @@ public class StorageEngine implements IService {
}
/** delete all data of storage groups' timeseries. */
+ @TestOnly
public synchronized boolean deleteAll() {
logger.info("Start deleting all storage groups' timeseries");
syncCloseAllProcessor();
for (PartialPath storageGroup :
IoTDB.metaManager.getAllStorageGroupPaths()) {
this.deleteAllDataFilesInOneStorageGroup(storageGroup);
}
+ processorMap.clear();
return true;
}
@@ -874,7 +920,7 @@ public class StorageEngine implements IService {
throws LoadFileException, StorageEngineException, MetadataException {
Set<String> deviceSet = newTsFileResource.getDevices();
if (deviceSet == null || deviceSet.isEmpty()) {
- throw new StorageEngineException("Can not get the corresponding storage
group.");
+ throw new StorageEngineException("The TsFile is empty, cannot be
loaded.");
}
String device = deviceSet.iterator().next();
PartialPath devicePath = new PartialPath(device);
@@ -890,13 +936,17 @@ public class StorageEngine implements IService {
public boolean deleteTsfile(File deletedTsfile)
throws StorageEngineException, IllegalPathException {
- return getProcessorDirectly(new
PartialPath(getSgByEngineFile(deletedTsfile, true)))
+ return getProcessorDirectly(
+ new PartialPath(getSgByEngineFile(deletedTsfile, true)),
+ getVirtualSgIdByEngineFile(deletedTsfile, true))
.deleteTsfile(deletedTsfile);
}
public boolean unloadTsfile(File tsfileToBeUnloaded, File targetDir)
throws StorageEngineException, IllegalPathException {
- return getProcessorDirectly(new
PartialPath(getSgByEngineFile(tsfileToBeUnloaded, true)))
+ return getProcessorDirectly(
+ new PartialPath(getSgByEngineFile(tsfileToBeUnloaded, true)),
+ getVirtualSgIdByEngineFile(tsfileToBeUnloaded, true))
.unloadTsfile(tsfileToBeUnloaded, targetDir);
}
@@ -932,6 +982,38 @@ public class StorageEngine implements IService {
}
}
+ /**
+ * The internal file means that the file is in the engine, which is
different from those external
+ * files which are not loaded.
+ *
+ * @param file internal file
+ * @param needCheck check if the tsfile is an internal TsFile. If you make
sure it is inside, no
+ * need to check
+ * @return virtual storage group partition id
+ * @throws IllegalPathException throw if tsfile is not an internal TsFile
+ */
+ public int getVirtualSgIdByEngineFile(File file, boolean needCheck) throws
IllegalPathException {
+ if (needCheck) {
+ File dataDir =
+
file.getParentFile().getParentFile().getParentFile().getParentFile().getParentFile();
+ if (dataDir.exists()) {
+ String[] dataDirs =
IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dir : dataDirs) {
+ try {
+ if (Files.isSameFile(Paths.get(dir), dataDir.toPath())) {
+ return
Integer.parseInt(file.getParentFile().getParentFile().getName());
+ }
+ } catch (IOException e) {
+ throw new IllegalPathException(file.getAbsolutePath(),
e.getMessage());
+ }
+ }
+ }
+ throw new IllegalPathException(file.getAbsolutePath(), "it's not an
internal tsfile.");
+ } else {
+ return Integer.parseInt(file.getParentFile().getParentFile().getName());
+ }
+ }
+
/** @return TsFiles (seq or unseq) grouped by their storage group and
partition number. */
public Map<PartialPath, Map<Long, List<TsFileResource>>>
getAllClosedStorageGroupTsFile() {
Map<PartialPath, Map<Long, List<TsFileResource>>> ret = new HashMap<>();
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/StorageGroupManager.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/StorageGroupManager.java
index cdd0cb276a..89234fb9d3 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/StorageGroupManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/StorageGroupManager.java
@@ -185,6 +185,42 @@ public class StorageGroupManager {
return processor;
}
+ /**
+ * get processor from virtualStorageGroupId
+ *
+ * @param virtualStorageGroupId virtual storage group id
+ * @return virtual storage group processor
+ */
+ @SuppressWarnings("java:S2445")
+ // actually storageGroupMNode is a unique object on the mtree, synchronize
it is reasonable
+ public VirtualStorageGroupProcessor getProcessor(
+ int virtualStorageGroupId, IStorageGroupMNode storageGroupMNode)
+ throws StorageGroupProcessorException, StorageEngineException {
+ VirtualStorageGroupProcessor processor =
virtualStorageGroupProcessor[virtualStorageGroupId];
+ if (processor == null) {
+ // if finish recover
+ if (isVsgReady[virtualStorageGroupId].get()) {
+ synchronized (storageGroupMNode) {
+ processor = virtualStorageGroupProcessor[virtualStorageGroupId];
+ if (processor == null) {
+ processor =
+ StorageEngine.getInstance()
+ .buildNewStorageGroupProcessor(
+ storageGroupMNode.getPartialPath(),
+ storageGroupMNode,
+ String.valueOf(virtualStorageGroupId));
+ virtualStorageGroupProcessor[virtualStorageGroupId] = processor;
+ }
+ }
+ } else {
+ // not finished recover, refuse the request
+ throw new StorageGroupNotReadyException(
+ storageGroupMNode.getFullPath(),
TSStatusCode.STORAGE_GROUP_NOT_READY.getStatusCode());
+ }
+ }
+
+ return processor;
+ }
/**
* async recover all virtual storage groups in this logical storage group
