You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2022/04/10 06:06:31 UTC
[iotdb] 01/01: Rename vsg to dataRegion
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch mv_vsg_dataregion
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8663731ba74647c61f3b98e861b6e3becc2cb1ce
Author: HTHou <hh...@outlook.com>
AuthorDate: Sun Apr 10 14:06:03 2022 +0800
Rename vsg to dataRegion
---
.../iotdb/cluster/ClusterFileFlushPolicy.java | 11 +-
.../iotdb/cluster/metadata/CSchemaProcessor.java | 8 +-
.../partition/slot/SlotTimePartitionFilter.java | 2 +-
.../iotdb/cluster/query/ClusterPlanExecutor.java | 2 +-
.../cluster/server/member/DataGroupMember.java | 2 +-
.../cluster/log/applier/DataLogApplierTest.java | 10 +-
.../cluster/log/snapshot/FileSnapshotTest.java | 14 +-
.../log/snapshot/PartitionedSnapshotTest.java | 6 +-
.../cluster/log/snapshot/PullSnapshotTaskTest.java | 4 +-
.../cluster/server/member/DataGroupMemberTest.java | 4 +-
.../cluster/server/member/MetaGroupMemberTest.java | 4 +-
.../iotdb/db/integration/IoTDBCompleteIT.java | 3 +-
.../db/integration/IoTDBLoadExternalTsfileIT.java | 6 +-
.../db/integration/IoTDBNewTsFileCompactionIT.java | 7 +-
.../aligned/IoTDBLoadExternalAlignedTsFileIT.java | 6 +-
.../db/integration/env/StandaloneEnvConfig.java | 2 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 12 +-
.../org/apache/iotdb/db/conf/IoTDBConfigCheck.java | 10 +-
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 4 +-
.../apache/iotdb/db/consensus/ConsensusImpl.java | 4 +-
.../statemachine/DataRegionStateMachine.java | 2 +-
.../org/apache/iotdb/db/engine/StorageEngine.java | 84 ++++------
.../apache/iotdb/db/engine/StorageEngineV2.java | 82 +++++-----
.../db/engine/compaction/CompactionScheduler.java | 6 +-
.../db/engine/compaction/TsFileIdentifier.java | 18 +--
.../AbstractInnerSpaceCompactionSelector.java | 6 +-
.../sizetiered/SizeTieredCompactionSelector.java | 2 +-
.../compaction/task/CompactionRecoverManager.java | 25 +--
.../iotdb/db/engine/flush/TsFileFlushPolicy.java | 14 +-
...lStorageGroupProcessor.java => DataRegion.java} | 121 ++++++--------
.../db/engine/storagegroup/StorageGroupInfo.java | 18 +--
.../db/engine/storagegroup/TsFileManager.java | 15 +-
.../db/engine/storagegroup/TsFileProcessor.java | 2 +-
.../db/engine/storagegroup/TsFileResource.java | 4 +-
.../HashVirtualPartitioner.java | 4 +-
.../StorageGroupManager.java | 175 +++++++++------------
.../VirtualPartitioner.java | 2 +-
.../apache/iotdb/db/metadata/tag/TagManager.java | 10 +-
.../apache/iotdb/db/mpp/execution/DataDriver.java | 4 +-
.../iotdb/db/mpp/execution/DataDriverContext.java | 8 +-
.../db/mpp/execution/FragmentInstanceManager.java | 4 +-
.../db/mpp/sql/planner/LocalExecutionPlanner.java | 4 +-
.../apache/iotdb/db/qp/executor/IPlanExecutor.java | 2 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 2 +-
.../iotdb/db/qp/physical/crud/DeletePlan.java | 2 +-
.../db/qp/physical/sys/DeleteTimeSeriesPlan.java | 2 +-
.../db/query/control/QueryResourceManager.java | 12 +-
.../groupby/GroupByWithValueFilterDataSet.java | 11 +-
.../groupby/GroupByWithoutValueFilterDataSet.java | 11 +-
.../db/query/executor/AggregationExecutor.java | 20 ++-
.../iotdb/db/query/executor/FillQueryExecutor.java | 10 +-
.../iotdb/db/query/executor/LastQueryExecutor.java | 10 +-
.../db/query/executor/RawDataQueryExecutor.java | 20 ++-
.../query/timegenerator/ServerTimeGenerator.java | 10 +-
.../db/sync/sender/manage/ISyncFileManager.java | 4 +-
.../db/sync/sender/manage/SyncFileManager.java | 12 +-
.../iotdb/db/sync/sender/transfer/SyncClient.java | 8 +-
.../db/tools/virtualsg/DeviceMappingViewer.java | 2 +-
.../iotdb/db/writelog/recover/LogReplayer.java | 12 +-
.../writelog/recover/TsFileRecoverPerformer.java | 10 +-
.../db/engine/compaction/TsFileIdentifierUT.java | 8 +-
.../cross/RewriteCrossSpaceCompactionTest.java | 10 +-
.../recover/SizeTieredCompactionRecoverTest.java | 4 +-
.../engine/modification/DeletionFileNodeTest.java | 24 ++-
.../storagegroup/StorageGroupProcessorTest.java | 4 +-
.../iotdb/db/engine/storagegroup/TTLTest.java | 54 +++----
.../HashVirtualPartitionerTest.java | 2 +-
.../db/metadata/idtable/IDTableFlushTimeTest.java | 10 +-
.../iotdb/db/mpp/execution/DataDriverTest.java | 4 +-
.../db/sync/receiver/load/FileLoaderTest.java | 23 ++-
.../recover/SyncReceiverLogAnalyzerTest.java | 5 +-
.../apache/iotdb/tsfile/utils/FilePathUtils.java | 2 +-
.../iotdb/tsfile/utils/FilePathUtilsTest.java | 10 +-
73 files changed, 467 insertions(+), 584 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterFileFlushPolicy.java b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterFileFlushPolicy.java
index 81ef9edc23..c642632206 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterFileFlushPolicy.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterFileFlushPolicy.java
@@ -21,8 +21,8 @@ package org.apache.iotdb.cluster;
import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,10 +56,7 @@ public class ClusterFileFlushPolicy implements TsFileFlushPolicy {
}
@Override
- public void apply(
- VirtualStorageGroupProcessor virtualStorageGroupProcessor,
- TsFileProcessor processor,
- boolean isSeq) {
+ public void apply(DataRegion dataRegion, TsFileProcessor processor, boolean isSeq) {
logger.info(
"The memtable size reaches the threshold, async flush it to tsfile: {}",
processor.getTsFileResource().getTsFile().getAbsolutePath());
@@ -70,9 +67,7 @@ public class ClusterFileFlushPolicy implements TsFileFlushPolicy {
closePartitionExecutor.submit(
() ->
metaGroupMember.closePartition(
- virtualStorageGroupProcessor.getVirtualStorageGroupId(),
- processor.getTimeRangeId(),
- isSeq));
+ dataRegion.getDataRegionId(), processor.getTimeRangeId(), isSeq));
}
// flush the memtable anyway to avoid the insertion trigger the policy again
processor.asyncFlush();
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CSchemaProcessor.java b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CSchemaProcessor.java
index cccf1bfe37..ac9e694d29 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CSchemaProcessor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CSchemaProcessor.java
@@ -41,7 +41,7 @@ import org.apache.iotdb.cluster.server.member.MetaGroupMember;
import org.apache.iotdb.cluster.utils.ClusterQueryUtils;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
@@ -286,9 +286,9 @@ public class CSchemaProcessor extends LocalSchemaProcessor {
}
/**
- * the {@link org.apache.iotdb.db.writelog.recover.LogReplayer#replayLogs(Supplier,
- * VirtualStorageGroupProcessor)} will call this to get schema after restart we should retry to
- * get schema util we get the schema.
+ * the {@link org.apache.iotdb.db.writelog.recover.LogReplayer#replayLogs(Supplier, DataRegion)}
+ * will call this to get schema after restart we should retry to get schema util we get the
+ * schema.
*
* @param deviceId the device id.
* @param measurements the measurements.
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java
index 7a7825f236..5bd42bb8fb 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/partition/slot/SlotTimePartitionFilter.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.cluster.partition.slot;
import org.apache.iotdb.cluster.config.ClusterConstant;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import java.util.Objects;
import java.util.Set;
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java
index 8b149bba22..0fbd5d2820 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/ClusterPlanExecutor.java
@@ -38,7 +38,7 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
index 47f88b6cab..c94a5afc0a 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
@@ -77,7 +77,7 @@ import org.apache.iotdb.commons.service.JMXService;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java
index 8fa0ad7be0..36e389bf60 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/applier/DataLogApplierTest.java
@@ -53,8 +53,8 @@ import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -407,13 +407,13 @@ public class DataLogApplierTest extends IoTDBTest {
@Test
public void testApplyCloseFile() throws IoTDBException {
- VirtualStorageGroupProcessor virtualStorageGroupProcessor =
+ DataRegion dataRegion =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
- TestCase.assertFalse(virtualStorageGroupProcessor.getWorkSequenceTsFileProcessors().isEmpty());
+ TestCase.assertFalse(dataRegion.getWorkSequenceTsFileProcessors().isEmpty());
CloseFileLog closeFileLog = new CloseFileLog(TestUtils.getTestSg(0), 0, true);
applier.apply(closeFileLog);
- TestCase.assertTrue(virtualStorageGroupProcessor.getWorkSequenceTsFileProcessors().isEmpty());
+ TestCase.assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty());
}
@Test
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/FileSnapshotTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/FileSnapshotTest.java
index d43258b3c3..36e0d9ecc8 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/FileSnapshotTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/FileSnapshotTest.java
@@ -27,8 +27,8 @@ import org.apache.iotdb.cluster.partition.slot.SlotManager.SlotStatus;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.LoadFileException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
@@ -117,7 +117,7 @@ public class FileSnapshotTest extends DataSnapshotTest {
assertTrue(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(9, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
@@ -161,7 +161,7 @@ public class FileSnapshotTest extends DataSnapshotTest {
assertTrue(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(9, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
@@ -211,7 +211,7 @@ public class FileSnapshotTest extends DataSnapshotTest {
assertTrue(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(9, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
@@ -251,7 +251,7 @@ public class FileSnapshotTest extends DataSnapshotTest {
defaultInstaller.install(snapshotMap, false);
for (int j = 0; j < 10; j++) {
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(j)));
assertEquals(9, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
@@ -276,7 +276,7 @@ public class FileSnapshotTest extends DataSnapshotTest {
timeseriesSchemas.add(TestUtils.getTestTimeSeriesSchema(0, i));
}
for (int i = 0; i < 5; i++) {
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
TsFileResource resource = tsFileResources.get(i);
String pathWithoutHardlinkSuffix =
@@ -297,7 +297,7 @@ public class FileSnapshotTest extends DataSnapshotTest {
assertTrue(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(10, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshotTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshotTest.java
index 77ebfcd9d5..dea05a1dbd 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshotTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshotTest.java
@@ -23,8 +23,8 @@ import org.apache.iotdb.cluster.common.TestUtils;
import org.apache.iotdb.cluster.exception.SnapshotInstallationException;
import org.apache.iotdb.cluster.partition.slot.SlotManager.SlotStatus;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.path.PartialPath;
@@ -110,7 +110,7 @@ public class PartitionedSnapshotTest extends DataSnapshotTest {
assertTrue(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(9, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
@@ -187,7 +187,7 @@ public class PartitionedSnapshotTest extends DataSnapshotTest {
assertFalse(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(-1, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PullSnapshotTaskTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PullSnapshotTaskTest.java
index 603f85f817..2bf2c8da52 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PullSnapshotTaskTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/log/snapshot/PullSnapshotTaskTest.java
@@ -37,8 +37,8 @@ import org.apache.iotdb.cluster.utils.IOUtils;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -292,7 +292,7 @@ public class PullSnapshotTaskTest extends DataSnapshotTest {
assertTrue(
IoTDB.schemaProcessor.isPathExist(new PartialPath(timeseriesSchema.getFullPath())));
}
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertEquals(9, processor.getPartitionMaxFileVersions(0));
List<TsFileResource> loadedFiles = processor.getSequenceFileList();
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
index c82800f203..4eb9057291 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/DataGroupMemberTest.java
@@ -64,8 +64,8 @@ import org.apache.iotdb.cluster.utils.Constants;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.TriggerExecutionException;
import org.apache.iotdb.db.exception.WriteProcessException;
@@ -502,7 +502,7 @@ public class DataGroupMemberTest extends BaseMember {
snapshot.addFile(tsFileResource, TestUtils.getNode(0), true);
// create a local resource1
- VirtualStorageGroupProcessor processor;
+ DataRegion processor;
while (true) {
try {
processor =
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
index da1cb4bf14..7fa84ef91a 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
@@ -84,7 +84,7 @@ import org.apache.iotdb.db.auth.authorizer.LocalFileAuthorizer;
import org.apache.iotdb.db.auth.entity.Role;
import org.apache.iotdb.db.auth.entity.User;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -613,7 +613,7 @@ public class MetaGroupMemberTest extends BaseMember {
ExecutorService testThreadPool = Executors.newFixedThreadPool(4);
assertTrue(testMetaMember.closePartition(TestUtils.getTestSg(0), 0, true));
- VirtualStorageGroupProcessor processor =
+ DataRegion processor =
StorageEngine.getInstance().getProcessor(new PartialPath(TestUtils.getTestSg(0)));
assertTrue(processor.getWorkSequenceTsFileProcessors().isEmpty());
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBCompleteIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBCompleteIT.java
index e30830cda6..7e69de17c2 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBCompleteIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBCompleteIT.java
@@ -44,8 +44,7 @@ public class IoTDBCompleteIT {
@Before
public void setUp() throws InterruptedException {
// test different partition
- prevVirtualStorageGroupNum =
- IoTDBDescriptor.getInstance().getConfig().getVirtualStorageGroupNum();
+ prevVirtualStorageGroupNum = IoTDBDescriptor.getInstance().getConfig().getDataRegionNum();
ConfigFactory.getConfig().setVirtualStorageGroupNum(16);
EnvFactory.getEnv().initBeforeClass();
}
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 f415e2503f..b636f2f697 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
@@ -140,8 +140,8 @@ public class IoTDBLoadExternalTsfileIT {
@Before
public void setUp() throws Exception {
- prevVirtualPartitionNum = IoTDBDescriptor.getInstance().getConfig().getVirtualStorageGroupNum();
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(1);
+ prevVirtualPartitionNum = IoTDBDescriptor.getInstance().getConfig().getDataRegionNum();
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(1);
prevCompactionThread =
IoTDBDescriptor.getInstance().getConfig().getConcurrentCompactionThread();
EnvironmentUtils.envSetUp();
@@ -153,7 +153,7 @@ public class IoTDBLoadExternalTsfileIT {
public void tearDown() throws Exception {
EnvironmentUtils.cleanEnv();
IoTDBDescriptor.getInstance().getConfig().setConcurrentCompactionThread(prevCompactionThread);
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(prevVirtualPartitionNum);
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(prevVirtualPartitionNum);
}
@Test
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 59734ed299..9afcd388ba 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
@@ -21,8 +21,8 @@ package org.apache.iotdb.db.integration;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.compaction.CompactionTaskManager;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -1062,9 +1062,8 @@ public class IoTDBNewTsFileCompactionIT {
/** wait until merge is finished */
private boolean waitForMergeFinish() throws StorageEngineException, InterruptedException {
- VirtualStorageGroupProcessor virtualStorageGroupProcessor =
- StorageEngine.getInstance().getProcessor(storageGroupPath);
- TsFileManager resourceManager = virtualStorageGroupProcessor.getTsFileResourceManager();
+ DataRegion dataRegion = StorageEngine.getInstance().getProcessor(storageGroupPath);
+ TsFileManager resourceManager = dataRegion.getTsFileResourceManager();
long startTime = System.nanoTime();
TimeUnit.MILLISECONDS.sleep(500);
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBLoadExternalAlignedTsFileIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBLoadExternalAlignedTsFileIT.java
index 5b0715e41c..feea79ab9e 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBLoadExternalAlignedTsFileIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/aligned/IoTDBLoadExternalAlignedTsFileIT.java
@@ -138,8 +138,8 @@ public class IoTDBLoadExternalAlignedTsFileIT {
@Before
public void setUp() throws Exception {
- prevVirtualPartitionNum = IoTDBDescriptor.getInstance().getConfig().getVirtualStorageGroupNum();
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(1);
+ prevVirtualPartitionNum = IoTDBDescriptor.getInstance().getConfig().getDataRegionNum();
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(1);
prevCompactionThread =
IoTDBDescriptor.getInstance().getConfig().getConcurrentCompactionThread();
EnvironmentUtils.envSetUp();
@@ -151,7 +151,7 @@ public class IoTDBLoadExternalAlignedTsFileIT {
public void tearDown() throws Exception {
EnvironmentUtils.cleanEnv();
IoTDBDescriptor.getInstance().getConfig().setConcurrentCompactionThread(prevCompactionThread);
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(prevVirtualPartitionNum);
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(prevVirtualPartitionNum);
}
@Test
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/env/StandaloneEnvConfig.java b/integration/src/test/java/org/apache/iotdb/db/integration/env/StandaloneEnvConfig.java
index bdfbe7a4b5..e40f3412a9 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/env/StandaloneEnvConfig.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/env/StandaloneEnvConfig.java
@@ -46,7 +46,7 @@ public class StandaloneEnvConfig implements BaseConfig {
}
public BaseConfig setVirtualStorageGroupNum(int virtualStorageGroupNum) {
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(virtualStorageGroupNum);
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(virtualStorageGroupNum);
return this;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 54464410d9..fe321c3e49 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -796,8 +796,8 @@ public class IoTDBConfig {
/** the size of ioTaskQueue */
private int ioTaskQueueSizeForFlushing = 10;
- /** the number of virtual storage groups per user-defined storage group */
- private int virtualStorageGroupNum = 1;
+ /** the number of data regions per user-defined storage group */
+ private int dataRegionNum = 1;
/** the interval to log recover progress of each vsg when starting iotdb */
private long recoveryLogIntervalInMs = 5_000L;
@@ -2341,12 +2341,12 @@ public class IoTDBConfig {
this.defaultIndexWindowRange = defaultIndexWindowRange;
}
- public int getVirtualStorageGroupNum() {
- return virtualStorageGroupNum;
+ public int getDataRegionNum() {
+ return dataRegionNum;
}
- public void setVirtualStorageGroupNum(int virtualStorageGroupNum) {
- this.virtualStorageGroupNum = virtualStorageGroupNum;
+ public void setDataRegionNum(int dataRegionNum) {
+ this.dataRegionNum = dataRegionNum;
}
public long getRecoveryLogIntervalInMs() {
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java
index fbc1295d41..5d464b657a 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfigCheck.java
@@ -88,8 +88,8 @@ public class IoTDBConfigCheck {
private static String maxDegreeOfIndexNode =
String.valueOf(TSFileDescriptor.getInstance().getConfig().getMaxDegreeOfIndexNode());
- private static final String VIRTUAL_STORAGE_GROUP_NUM = "virtual_storage_group_num";
- private static String virtualStorageGroupNum = String.valueOf(config.getVirtualStorageGroupNum());
+ private static final String DATA_REGION_NUM = "virtual_storage_group_num";
+ private static String dataRegionNum = String.valueOf(config.getDataRegionNum());
private static final String ENABLE_ID_TABLE = "enable_id_table";
private static String enableIDTable = String.valueOf(config.isEnableIDTable());
@@ -155,7 +155,7 @@ public class IoTDBConfigCheck {
systemProperties.put(TAG_ATTRIBUTE_SIZE_STRING, tagAttributeTotalSize);
systemProperties.put(TAG_ATTRIBUTE_FLUSH_INTERVAL, tagAttributeFlushInterval);
systemProperties.put(MAX_DEGREE_OF_INDEX_STRING, maxDegreeOfIndexNode);
- systemProperties.put(VIRTUAL_STORAGE_GROUP_NUM, virtualStorageGroupNum);
+ systemProperties.put(DATA_REGION_NUM, dataRegionNum);
systemProperties.put(TIME_ENCODER_KEY, timeEncoderValue);
systemProperties.put(ENABLE_ID_TABLE, enableIDTable);
systemProperties.put(ENABLE_ID_TABLE_LOG_FILE, enableIdTableLogFile);
@@ -344,8 +344,8 @@ public class IoTDBConfigCheck {
throwException(MAX_DEGREE_OF_INDEX_STRING, maxDegreeOfIndexNode);
}
- if (!(properties.getProperty(VIRTUAL_STORAGE_GROUP_NUM).equals(virtualStorageGroupNum))) {
- throwException(VIRTUAL_STORAGE_GROUP_NUM, virtualStorageGroupNum);
+ if (!(properties.getProperty(DATA_REGION_NUM).equals(dataRegionNum))) {
+ throwException(DATA_REGION_NUM, dataRegionNum);
}
if (!(properties.getProperty(TIME_ENCODER_KEY).equals(timeEncoderValue))) {
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index ec955aa40a..cf77ed597a 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -686,10 +686,10 @@ public class IoTDBDescriptor {
properties.getProperty(
"slow_query_threshold", String.valueOf(conf.getSlowQueryThreshold()))));
- conf.setVirtualStorageGroupNum(
+ conf.setDataRegionNum(
Integer.parseInt(
properties.getProperty(
- "virtual_storage_group_num", String.valueOf(conf.getVirtualStorageGroupNum()))));
+ "virtual_storage_group_num", String.valueOf(conf.getDataRegionNum()))));
conf.setRecoveryLogIntervalInMs(
Long.parseLong(
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/ConsensusImpl.java b/server/src/main/java/org/apache/iotdb/db/consensus/ConsensusImpl.java
index 68476e75ec..972221609f 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/ConsensusImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/ConsensusImpl.java
@@ -28,7 +28,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.consensus.statemachine.DataRegionStateMachine;
import org.apache.iotdb.db.consensus.statemachine.SchemaRegionStateMachine;
-import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.StorageEngineV2;
import org.apache.iotdb.db.metadata.schemaregion.SchemaEngine;
import java.io.File;
@@ -60,7 +60,7 @@ public class ConsensusImpl {
SchemaEngine.getInstance().getSchemaRegion((SchemaRegionId) gid));
case DataRegion:
return new DataRegionStateMachine(
- StorageEngine.getInstance().getDataRegion((DataRegionId) gid));
+ StorageEngineV2.getInstance().getDataRegion((DataRegionId) gid));
}
throw new IllegalArgumentException(
String.format("Unexpected consensusGroup %s", gid));
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
index 06312cd13f..5f6e168f6c 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.consensus.statemachine;
import org.apache.iotdb.consensus.common.DataSet;
-import org.apache.iotdb.db.mpp.common.DataRegion;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.mpp.sql.planner.plan.FragmentInstance;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
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 b0551f5e99..304772bbab 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
@@ -20,7 +20,6 @@ package org.apache.iotdb.db.engine;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
-import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.exception.ShutdownException;
import org.apache.iotdb.commons.partition.TimePartitionSlot;
import org.apache.iotdb.commons.service.IService;
@@ -34,11 +33,11 @@ import org.apache.iotdb.db.engine.flush.CloseFileListener;
import org.apache.iotdb.db.engine.flush.FlushListener;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
-import org.apache.iotdb.db.engine.storagegroup.virtualSg.StorageGroupManager;
+import org.apache.iotdb.db.engine.storagegroup.dataregion.StorageGroupManager;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.LoadFileException;
import org.apache.iotdb.db.exception.StorageEngineException;
@@ -55,7 +54,6 @@ import org.apache.iotdb.db.metadata.idtable.entry.DeviceIDFactory;
import org.apache.iotdb.db.metadata.mnode.IMeasurementMNode;
import org.apache.iotdb.db.metadata.mnode.IStorageGroupMNode;
import org.apache.iotdb.db.metadata.path.PartialPath;
-import org.apache.iotdb.db.mpp.common.DataRegion;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
@@ -138,8 +136,6 @@ public class StorageEngine implements IService {
private List<CloseFileListener> customCloseFileListeners = new ArrayList<>();
private List<FlushListener> customFlushListeners = new ArrayList<>();
- private final Map<DataRegionId, DataRegion> regionMap = new ConcurrentHashMap<>();
-
private StorageEngine() {}
public static StorageEngine getInstance() {
@@ -470,8 +466,7 @@ public class StorageEngine implements IService {
* @param path storage group path
* @return storage group processor
*/
- public VirtualStorageGroupProcessor getProcessorDirectly(PartialPath path)
- throws StorageEngineException {
+ public DataRegion getProcessorDirectly(PartialPath path) throws StorageEngineException {
PartialPath storageGroupPath;
try {
IStorageGroupMNode storageGroupMNode = IoTDB.schemaProcessor.getStorageGroupNodeByPath(path);
@@ -488,7 +483,7 @@ public class StorageEngine implements IService {
* @param path device path
* @return storage group processor
*/
- public VirtualStorageGroupProcessor getProcessor(PartialPath path) throws StorageEngineException {
+ public DataRegion getProcessor(PartialPath path) throws StorageEngineException {
try {
IStorageGroupMNode storageGroupMNode = IoTDB.schemaProcessor.getStorageGroupNodeByPath(path);
return getStorageGroupProcessorByPath(path, storageGroupMNode);
@@ -508,9 +503,8 @@ public class StorageEngine implements IService {
for (PartialPath path : pathList) {
IStorageGroupMNode storageGroupMNode =
IoTDB.schemaProcessor.getStorageGroupNodeByPath(path);
- VirtualStorageGroupProcessor virtualStorageGroupProcessor =
- getStorageGroupProcessorByPath(path, storageGroupMNode);
- lockHolderList.add(virtualStorageGroupProcessor.getInsertWriteLockHolder());
+ DataRegion dataRegion = getStorageGroupProcessorByPath(path, storageGroupMNode);
+ lockHolderList.add(dataRegion.getInsertWriteLockHolder());
}
return lockHolderList;
} catch (StorageGroupProcessorException | MetadataException e) {
@@ -528,7 +522,7 @@ public class StorageEngine implements IService {
*/
@SuppressWarnings("java:S2445")
// actually storageGroupMNode is a unique object on the mtree, synchronize it is reasonable
- private VirtualStorageGroupProcessor getStorageGroupProcessorByPath(
+ private DataRegion getStorageGroupProcessorByPath(
PartialPath devicePath, IStorageGroupMNode storageGroupMNode)
throws StorageGroupProcessorException, StorageEngineException {
StorageGroupManager storageGroupManager = processorMap.get(storageGroupMNode.getPartialPath());
@@ -550,18 +544,18 @@ public class StorageEngine implements IService {
* @param virtualStorageGroupId virtual storage group id e.g. 1
* @param logicalStorageGroupName logical storage group name e.g. root.sg1
*/
- public VirtualStorageGroupProcessor buildNewStorageGroupProcessor(
+ public DataRegion buildNewStorageGroupProcessor(
PartialPath logicalStorageGroupName,
IStorageGroupMNode storageGroupMNode,
String virtualStorageGroupId)
throws StorageGroupProcessorException {
- VirtualStorageGroupProcessor processor;
+ DataRegion processor;
logger.info(
"construct a processor instance, the storage group is {}, Thread is {}",
logicalStorageGroupName,
Thread.currentThread().getId());
processor =
- new VirtualStorageGroupProcessor(
+ new DataRegion(
systemDir + File.separator + logicalStorageGroupName,
virtualStorageGroupId,
fileFlushPolicy,
@@ -594,9 +588,8 @@ public class StorageEngine implements IService {
}
}
- VirtualStorageGroupProcessor virtualStorageGroupProcessor =
- getProcessor(insertRowPlan.getDevicePath());
- getSeriesSchemas(insertRowPlan, virtualStorageGroupProcessor);
+ DataRegion dataRegion = getProcessor(insertRowPlan.getDevicePath());
+ getSeriesSchemas(insertRowPlan, dataRegion);
try {
insertRowPlan.transferType();
} catch (QueryProcessException e) {
@@ -604,7 +597,7 @@ public class StorageEngine implements IService {
}
try {
- virtualStorageGroupProcessor.insert(insertRowPlan);
+ dataRegion.insert(insertRowPlan);
} catch (WriteProcessException e) {
throw new StorageEngineException(e);
}
@@ -620,18 +613,17 @@ public class StorageEngine implements IService {
}
}
- VirtualStorageGroupProcessor virtualStorageGroupProcessor =
- getProcessor(insertRowsOfOneDevicePlan.getDevicePath());
+ DataRegion dataRegion = getProcessor(insertRowsOfOneDevicePlan.getDevicePath());
for (InsertRowPlan plan : insertRowsOfOneDevicePlan.getRowPlans()) {
plan.setMeasurementMNodes(new IMeasurementMNode[plan.getMeasurements().length]);
// check whether types are match
- getSeriesSchemas(plan, virtualStorageGroupProcessor);
+ getSeriesSchemas(plan, dataRegion);
}
// TODO monitor: update statistics
try {
- virtualStorageGroupProcessor.insert(insertRowsOfOneDevicePlan);
+ dataRegion.insert(insertRowsOfOneDevicePlan);
} catch (WriteProcessException e) {
throw new StorageEngineException(e);
}
@@ -649,9 +641,9 @@ public class StorageEngine implements IService {
throw new BatchProcessException(results);
}
}
- VirtualStorageGroupProcessor virtualStorageGroupProcessor;
+ DataRegion dataRegion;
try {
- virtualStorageGroupProcessor = getProcessor(insertTabletPlan.getDevicePath());
+ dataRegion = getProcessor(insertTabletPlan.getDevicePath());
} catch (StorageEngineException e) {
throw new StorageEngineException(
String.format(
@@ -660,8 +652,8 @@ public class StorageEngine implements IService {
e);
}
- getSeriesSchemas(insertTabletPlan, virtualStorageGroupProcessor);
- virtualStorageGroupProcessor.insertTablet(insertTabletPlan);
+ getSeriesSchemas(insertTabletPlan, dataRegion);
+ dataRegion.insertTablet(insertTabletPlan);
}
/** flush command Sync asyncCloseOneProcessor all file node processors. */
@@ -1042,37 +1034,36 @@ public class StorageEngine implements IService {
}
/** get all merge lock of the storage group processor related to the query */
- public Pair<
- List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- mergeLock(List<PartialPath> pathList) throws StorageEngineException {
- Map<VirtualStorageGroupProcessor, List<PartialPath>> map = new HashMap<>();
+ public Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> mergeLock(
+ List<PartialPath> pathList) throws StorageEngineException {
+ Map<DataRegion, List<PartialPath>> map = new HashMap<>();
for (PartialPath path : pathList) {
map.computeIfAbsent(getProcessor(path.getDevicePath()), key -> new ArrayList<>()).add(path);
}
- List<VirtualStorageGroupProcessor> list =
+ List<DataRegion> list =
map.keySet().stream()
- .sorted(Comparator.comparing(VirtualStorageGroupProcessor::getVirtualStorageGroupId))
+ .sorted(Comparator.comparing(DataRegion::getDataRegionId))
.collect(Collectors.toList());
- list.forEach(VirtualStorageGroupProcessor::readLock);
+ list.forEach(DataRegion::readLock);
return new Pair<>(list, map);
}
/** unlock all merge lock of the storage group processor related to the query */
- public void mergeUnLock(List<VirtualStorageGroupProcessor> list) {
- list.forEach(VirtualStorageGroupProcessor::readUnlock);
+ public void mergeUnLock(List<DataRegion> list) {
+ list.forEach(DataRegion::readUnlock);
}
/** @return virtual storage group name, like root.sg1/0 */
public String getStorageGroupPath(PartialPath path) throws StorageEngineException {
PartialPath deviceId = path.getDevicePath();
- VirtualStorageGroupProcessor storageGroupProcessor = getProcessor(deviceId);
+ DataRegion storageGroupProcessor = getProcessor(deviceId);
return storageGroupProcessor.getLogicalStorageGroupName()
+ File.separator
- + storageGroupProcessor.getVirtualStorageGroupId();
+ + storageGroupProcessor.getDataRegionId();
}
- protected void getSeriesSchemas(InsertPlan insertPlan, VirtualStorageGroupProcessor processor)
+ protected void getSeriesSchemas(InsertPlan insertPlan, DataRegion processor)
throws StorageEngineException, MetadataException {
try {
if (config.isEnableIDTable()) {
@@ -1087,17 +1078,6 @@ public class StorageEngine implements IService {
}
}
- // When registering a new region, the coordinator needs to register the corresponding region with
- // the local engine before adding the corresponding consensusGroup to the consensus layer
- // TODO implement it
- public DataRegion createDataRegion(DataRegionId regionId, String sg, long ttl) {
- return null;
- }
-
- public DataRegion getDataRegion(DataRegionId regionId) {
- return regionMap.get(regionId);
- }
-
static class InstanceHolder {
private static final StorageEngine INSTANCE = new StorageEngine();
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
index 538ae1ef67..31fe2b721d 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngineV2.java
@@ -34,8 +34,8 @@ import org.apache.iotdb.db.engine.flush.CloseFileListener;
import org.apache.iotdb.db.engine.flush.FlushListener;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
@@ -95,7 +95,7 @@ public class StorageEngineV2 implements IService {
FilePathUtils.regularizePath(config.getSystemDir()) + "storage_groups";
/** DataRegionId -> DataRegion */
- private final ConcurrentHashMap<ConsensusGroupId, VirtualStorageGroupProcessor> dataRegionMap =
+ private final ConcurrentHashMap<ConsensusGroupId, DataRegion> dataRegionMap =
new ConcurrentHashMap<>();
private AtomicBoolean isAllSgReady = new AtomicBoolean(false);
@@ -240,15 +240,14 @@ public class StorageEngineV2 implements IService {
continue;
}
ConsensusGroupId dataRegionId = new DataRegionId(Integer.parseInt(dataRegionDir.getName()));
- VirtualStorageGroupProcessor dataRegion =
- buildNewStorageGroupProcessor(sg, dataRegionDir.getName(), ttl);
+ DataRegion dataRegion = buildNewStorageGroupProcessor(sg, dataRegionDir.getName(), ttl);
dataRegionMap.putIfAbsent(dataRegionId, dataRegion);
}
}
}
private void asyncRecover(ExecutorService pool, List<Future<Void>> futures) {
- for (VirtualStorageGroupProcessor processor : dataRegionMap.values()) {
+ for (DataRegion processor : dataRegionMap.values()) {
Callable<Void> recoverVsgTask =
() -> {
processor.setReady(true);
@@ -289,7 +288,7 @@ public class StorageEngineV2 implements IService {
private void checkTTL() {
try {
- for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ for (DataRegion dataRegion : dataRegionMap.values()) {
if (dataRegion != null) {
dataRegion.checkFilesTTL();
}
@@ -342,7 +341,7 @@ public class StorageEngineV2 implements IService {
private void timedFlushSeqMemTable() {
try {
- for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ for (DataRegion dataRegion : dataRegionMap.values()) {
if (dataRegion != null) {
dataRegion.timedFlushSeqMemTable();
}
@@ -354,7 +353,7 @@ public class StorageEngineV2 implements IService {
private void timedFlushUnseqMemTable() {
try {
- for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ for (DataRegion dataRegion : dataRegionMap.values()) {
if (dataRegion != null) {
dataRegion.timedFlushUnseqMemTable();
}
@@ -366,7 +365,7 @@ public class StorageEngineV2 implements IService {
private void timedCloseTsFileProcessor() {
try {
- for (VirtualStorageGroupProcessor dataRegion : dataRegionMap.values()) {
+ for (DataRegion dataRegion : dataRegionMap.values()) {
if (dataRegion != null) {
dataRegion.timedCloseTsFileProcessor();
}
@@ -378,7 +377,7 @@ public class StorageEngineV2 implements IService {
@Override
public void stop() {
- for (VirtualStorageGroupProcessor vsg : dataRegionMap.values()) {
+ for (DataRegion vsg : dataRegionMap.values()) {
if (vsg != null) {
ThreadUtils.stopThreadPool(
vsg.getTimedCompactionScheduleTask(), ThreadName.COMPACTION_SCHEDULE);
@@ -403,12 +402,10 @@ public class StorageEngineV2 implements IService {
@Override
public void shutdown(long milliseconds) throws ShutdownException {
try {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor : dataRegionMap.values()) {
+ for (DataRegion dataRegion : dataRegionMap.values()) {
ThreadUtils.stopThreadPool(
- virtualStorageGroupProcessor.getTimedCompactionScheduleTask(),
- ThreadName.COMPACTION_SCHEDULE);
- ThreadUtils.stopThreadPool(
- virtualStorageGroupProcessor.getWALTrimScheduleTask(), ThreadName.WAL_TRIM);
+ dataRegion.getTimedCompactionScheduleTask(), ThreadName.COMPACTION_SCHEDULE);
+ ThreadUtils.stopThreadPool(dataRegion.getWALTrimScheduleTask(), ThreadName.WAL_TRIM);
}
forceCloseAllProcessor();
} catch (TsFileProcessorException e) {
@@ -458,16 +455,16 @@ public class StorageEngineV2 implements IService {
* @param virtualStorageGroupId virtual storage group id e.g. 1
* @param logicalStorageGroupName logical storage group name e.g. root.sg1
*/
- public VirtualStorageGroupProcessor buildNewStorageGroupProcessor(
+ public DataRegion buildNewStorageGroupProcessor(
String logicalStorageGroupName, String virtualStorageGroupId, long ttl)
throws StorageGroupProcessorException {
- VirtualStorageGroupProcessor processor;
+ DataRegion processor;
logger.info(
"construct a processor instance, the storage group is {}, Thread is {}",
logicalStorageGroupName,
Thread.currentThread().getId());
processor =
- new VirtualStorageGroupProcessor(
+ new DataRegion(
systemDir + File.separator + logicalStorageGroupName,
virtualStorageGroupId,
fileFlushPolicy,
@@ -500,7 +497,7 @@ public class StorageEngineV2 implements IService {
}
}
- VirtualStorageGroupProcessor dataRegion = dataRegionMap.get(dataRegionId);
+ DataRegion dataRegion = dataRegionMap.get(dataRegionId);
try {
dataRegion.insert(insertRowNode);
@@ -522,32 +519,32 @@ public class StorageEngineV2 implements IService {
throw new BatchProcessException(results);
}
}
- VirtualStorageGroupProcessor dataRegion = dataRegionMap.get(dataRegionId);
+ DataRegion dataRegion = dataRegionMap.get(dataRegionId);
dataRegion.insertTablet(insertTabletNode);
}
/** flush command Sync asyncCloseOneProcessor all file node processors. */
public void syncCloseAllProcessor() {
logger.info("Start closing all storage group processor");
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor : dataRegionMap.values()) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ for (DataRegion dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
}
}
public void forceCloseAllProcessor() throws TsFileProcessorException {
logger.info("Start force closing all storage group processor");
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor : dataRegionMap.values()) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.forceCloseAllWorkingTsFileProcessors();
+ for (DataRegion dataRegion : dataRegionMap.values()) {
+ if (dataRegion != null) {
+ dataRegion.forceCloseAllWorkingTsFileProcessors();
}
}
}
public void setTTL(List<ConsensusGroupId> dataRegionIdList, long dataTTL) {
for (ConsensusGroupId dataRegionId : dataRegionIdList) {
- VirtualStorageGroupProcessor dataRegion = dataRegionMap.get(dataRegionId);
+ DataRegion dataRegion = dataRegionMap.get(dataRegionId);
if (dataRegion != null) {
dataRegion.setDataTTL(dataTTL);
}
@@ -566,23 +563,21 @@ public class StorageEngineV2 implements IService {
*/
public Map<String, List<Pair<Long, Boolean>>> getWorkingStorageGroupPartitions() {
Map<String, List<Pair<Long, Boolean>>> res = new ConcurrentHashMap<>();
- for (Entry<ConsensusGroupId, VirtualStorageGroupProcessor> entry : dataRegionMap.entrySet()) {
- VirtualStorageGroupProcessor virtualStorageGroupProcessor = entry.getValue();
- if (virtualStorageGroupProcessor != null) {
+ for (Entry<ConsensusGroupId, DataRegion> entry : dataRegionMap.entrySet()) {
+ DataRegion dataRegion = entry.getValue();
+ if (dataRegion != null) {
List<Pair<Long, Boolean>> partitionIdList = new ArrayList<>();
- for (TsFileProcessor tsFileProcessor :
- virtualStorageGroupProcessor.getWorkSequenceTsFileProcessors()) {
+ for (TsFileProcessor tsFileProcessor : dataRegion.getWorkSequenceTsFileProcessors()) {
Pair<Long, Boolean> tmpPair = new Pair<>(tsFileProcessor.getTimeRangeId(), true);
partitionIdList.add(tmpPair);
}
- for (TsFileProcessor tsFileProcessor :
- virtualStorageGroupProcessor.getWorkUnsequenceTsFileProcessors()) {
+ for (TsFileProcessor tsFileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
Pair<Long, Boolean> tmpPair = new Pair<>(tsFileProcessor.getTimeRangeId(), false);
partitionIdList.add(tmpPair);
}
- res.put(virtualStorageGroupProcessor.getStorageGroupPath(), partitionIdList);
+ res.put(dataRegion.getStorageGroupPath(), partitionIdList);
}
}
@@ -611,19 +606,14 @@ public class StorageEngineV2 implements IService {
// When registering a new region, the coordinator needs to register the corresponding region with
// the local engine before adding the corresponding consensusGroup to the consensus layer
- public VirtualStorageGroupProcessor createDataRegion(DataRegionId regionId, String sg, long ttl)
- throws StorageEngineException {
- try {
- VirtualStorageGroupProcessor dataRegion =
- buildNewStorageGroupProcessor(sg, regionId.toString(), ttl);
- dataRegionMap.put(regionId, dataRegion);
- } catch (StorageGroupProcessorException e) {
- throw new StorageEngineException(e);
- }
- return null;
+ public DataRegion createDataRegion(DataRegionId regionId, String sg, long ttl)
+ throws StorageGroupProcessorException {
+ DataRegion dataRegion = buildNewStorageGroupProcessor(sg, regionId.toString(), ttl);
+ dataRegionMap.put(regionId, dataRegion);
+ return dataRegion;
}
- public VirtualStorageGroupProcessor getDataRegion(DataRegionId regionId) {
+ public DataRegion getDataRegion(DataRegionId regionId) {
return dataRegionMap.get(regionId);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionScheduler.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionScheduler.java
index 62dc5f424e..d9b641f472 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionScheduler.java
@@ -48,21 +48,21 @@ public class CompactionScheduler {
}
tryToSubmitCrossSpaceCompactionTask(
tsFileManager.getStorageGroupName(),
- tsFileManager.getVirtualStorageGroup(),
+ tsFileManager.getDataRegion(),
tsFileManager.getStorageGroupDir(),
timePartition,
tsFileManager,
new CrossSpaceCompactionTaskFactory());
tryToSubmitInnerSpaceCompactionTask(
tsFileManager.getStorageGroupName(),
- tsFileManager.getVirtualStorageGroup(),
+ tsFileManager.getDataRegion(),
timePartition,
tsFileManager,
true,
new InnerSpaceCompactionTaskFactory());
tryToSubmitInnerSpaceCompactionTask(
tsFileManager.getStorageGroupName(),
- tsFileManager.getVirtualStorageGroup(),
+ tsFileManager.getDataRegion(),
timePartition,
tsFileManager,
false,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java
index 503eafc570..f88597544c 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java
@@ -31,7 +31,7 @@ import java.io.File;
*/
public class TsFileIdentifier {
private final String logicalStorageGroupName;
- private final String virtualStorageGroupId;
+ private final String dataRegionId;
private final String timePartitionId;
private final boolean sequence;
private final String filename;
@@ -57,12 +57,12 @@ public class TsFileIdentifier {
private TsFileIdentifier(
String logicalStorageGroupName,
- String virtualStorageGroupId,
+ String dataRegionId,
String timePartitionId,
boolean sequence,
String filename) {
this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ this.dataRegionId = dataRegionId;
this.timePartitionId = timePartitionId;
this.sequence = sequence;
this.filename = filename;
@@ -150,7 +150,7 @@ public class TsFileIdentifier {
INFO_SEPARATOR,
logicalStorageGroupName,
INFO_SEPARATOR,
- virtualStorageGroupId,
+ dataRegionId,
INFO_SEPARATOR,
timePartitionId,
INFO_SEPARATOR,
@@ -165,7 +165,7 @@ public class TsFileIdentifier {
TsFileIdentifier otherInfo = (TsFileIdentifier) other;
return otherInfo.sequence == this.sequence
&& otherInfo.logicalStorageGroupName.equals(this.logicalStorageGroupName)
- && otherInfo.virtualStorageGroupId.equals(this.virtualStorageGroupId)
+ && otherInfo.dataRegionId.equals(this.dataRegionId)
&& otherInfo.timePartitionId.equals(this.timePartitionId)
&& otherInfo.filename.equals(this.filename);
}
@@ -181,7 +181,7 @@ public class TsFileIdentifier {
+ File.separator
+ logicalStorageGroupName
+ File.separator
- + virtualStorageGroupId
+ + dataRegionId
+ File.separator
+ timePartitionId
+ File.separator
@@ -204,7 +204,7 @@ public class TsFileIdentifier {
+ File.separator
+ logicalStorageGroupName
+ File.separator
- + virtualStorageGroupId
+ + dataRegionId
+ File.separator
+ timePartitionId
+ File.separator
@@ -215,8 +215,8 @@ public class TsFileIdentifier {
return logicalStorageGroupName;
}
- public String getVirtualStorageGroupId() {
- return virtualStorageGroupId;
+ public String getDataRegionId() {
+ return dataRegionId;
}
public String getTimePartitionId() {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionSelector.java
index 405e9c901d..cb7ec1029a 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionSelector.java
@@ -26,7 +26,7 @@ import java.util.List;
public abstract class AbstractInnerSpaceCompactionSelector extends AbstractCompactionSelector {
protected String logicalStorageGroupName;
- protected String virtualStorageGroupName;
+ protected String dataRegionName;
protected long timePartition;
protected List<TsFileResource> tsFileResources;
protected boolean sequence;
@@ -35,13 +35,13 @@ public abstract class AbstractInnerSpaceCompactionSelector extends AbstractCompa
public AbstractInnerSpaceCompactionSelector(
String logicalStorageGroupName,
- String virtualStorageGroupName,
+ String dataRegionName,
long timePartition,
TsFileManager tsFileManager,
boolean sequence,
InnerSpaceCompactionTaskFactory taskFactory) {
this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupName = virtualStorageGroupName;
+ this.dataRegionName = dataRegionName;
this.timePartition = timePartition;
this.tsFileManager = tsFileManager;
this.sequence = sequence;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java
index 8783aec004..9ef486a973 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java
@@ -172,7 +172,7 @@ public class SizeTieredCompactionSelector extends AbstractInnerSpaceCompactionSe
AbstractCompactionTask compactionTask =
taskFactory.createTask(
logicalStorageGroupName,
- virtualStorageGroupName,
+ dataRegionName,
timePartition,
tsFileManager,
selectedFileList,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverManager.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverManager.java
index 9d14c0482e..fe46eb0c71 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverManager.java
@@ -44,13 +44,13 @@ public class CompactionRecoverManager {
LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
private final TsFileManager tsFileManager;
private final String logicalStorageGroupName;
- private final String virtualStorageGroupId;
+ private final String dataRegionId;
public CompactionRecoverManager(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ TsFileManager tsFileManager, String logicalStorageGroupName, String dataRegionId) {
this.tsFileManager = tsFileManager;
this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ this.dataRegionId = dataRegionId;
}
public void recoverInnerSpaceCompaction(boolean isSequence) {
@@ -74,12 +74,7 @@ public class CompactionRecoverManager {
}
for (String dir : dirs) {
File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
+ new File(dir + File.separator + logicalStorageGroupName + File.separator + dataRegionId);
if (!storageGroupDir.exists()) {
return;
}
@@ -97,11 +92,7 @@ public class CompactionRecoverManager {
for (File compactionLog : compactionLogs) {
logger.info("Calling compaction recover task.");
new CompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- tsFileManager,
- compactionLog,
- isInnerSpace)
+ logicalStorageGroupName, dataRegionId, tsFileManager, compactionLog, isInnerSpace)
.doCompaction();
}
}
@@ -120,11 +111,7 @@ public class CompactionRecoverManager {
if (logFileFromOld.exists()) {
logger.info("Calling compaction task to recover from previous version.");
new CompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- tsFileManager,
- logFileFromOld,
- isInnerSpace)
+ logicalStorageGroupName, dataRegionId, tsFileManager, logFileFromOld, isInnerSpace)
.doCompaction();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/TsFileFlushPolicy.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/TsFileFlushPolicy.java
index d71df87fd8..76ee02a1ad 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/TsFileFlushPolicy.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/TsFileFlushPolicy.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.db.engine.flush;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
/**
* TsFileFlushPolicy is applied when a TsFileProcessor is full after insertion. For standalone
@@ -29,20 +29,14 @@ import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
*/
public interface TsFileFlushPolicy {
- void apply(
- VirtualStorageGroupProcessor virtualStorageGroupProcessor,
- TsFileProcessor processor,
- boolean isSeq);
+ void apply(DataRegion dataRegion, TsFileProcessor processor, boolean isSeq);
class DirectFlushPolicy implements TsFileFlushPolicy {
@Override
- public void apply(
- VirtualStorageGroupProcessor virtualStorageGroupProcessor,
- TsFileProcessor tsFileProcessor,
- boolean isSeq) {
+ public void apply(DataRegion dataRegion, TsFileProcessor tsFileProcessor, boolean isSeq) {
if (tsFileProcessor.shouldClose()) {
- virtualStorageGroupProcessor.asyncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
+ dataRegion.asyncCloseOneTsFileProcessor(isSeq, tsFileProcessor);
} else {
tsFileProcessor.asyncFlush();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
rename to server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
index 3b95fad455..a13e16c155 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
@@ -128,7 +128,7 @@ import static org.apache.iotdb.db.engine.storagegroup.TsFileResource.TEMP_SUFFIX
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
/**
- * For sequence data, a StorageGroupProcessor has some TsFileProcessors, in which there is only one
+ * For sequence data, a DataRegion has some TsFileProcessors, in which there is only one
* TsFileProcessor in the working status. <br>
*
* <p>There are two situations to set the working TsFileProcessor to closing status:<br>
@@ -147,7 +147,7 @@ import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFF
* <p>When a TsFileProcessor is closed, the closeUnsealedTsFileProcessorCallBack() method will be
* called as a callback.
*/
-public class VirtualStorageGroupProcessor {
+public class DataRegion {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final Logger DEBUG_LOGGER = LoggerFactory.getLogger("QUERY_DEBUG");
@@ -158,7 +158,7 @@ public class VirtualStorageGroupProcessor {
*/
private static final int MERGE_MOD_START_VERSION_NUM = 1;
- private static final Logger logger = LoggerFactory.getLogger(VirtualStorageGroupProcessor.class);
+ private static final Logger logger = LoggerFactory.getLogger(DataRegion.class);
/** indicating the file to be loaded overlap with some files. */
private static final int POS_OVERLAP = -3;
@@ -202,8 +202,8 @@ public class VirtualStorageGroupProcessor {
private AtomicBoolean isSettling = new AtomicBoolean();
- /** virtual storage group id */
- private String virtualStorageGroupId;
+ /** data region id */
+ private String dataRegionId;
/** logical storage group name */
private String logicalStorageGroupName;
/** storage group system directory */
@@ -291,7 +291,7 @@ public class VirtualStorageGroupProcessor {
logger.error(
"getDirectByteBuffer occurs error while waiting for DirectByteBuffer" + "group {}-{}",
logicalStorageGroupName,
- virtualStorageGroupId,
+ dataRegionId,
e);
}
logger.info(
@@ -369,24 +369,23 @@ public class VirtualStorageGroupProcessor {
* constrcut a storage group processor
*
* @param systemDir system dir path
- * @param virtualStorageGroupId virtual storage group id e.g. 1
+ * @param dataRegionId data region id e.g. 1
* @param fileFlushPolicy file flush policy
* @param logicalStorageGroupName logical storage group name e.g. root.sg1
*/
- public VirtualStorageGroupProcessor(
+ public DataRegion(
String systemDir,
- String virtualStorageGroupId,
+ String dataRegionId,
TsFileFlushPolicy fileFlushPolicy,
String logicalStorageGroupName)
throws StorageGroupProcessorException {
- this.virtualStorageGroupId = virtualStorageGroupId;
+ this.dataRegionId = dataRegionId;
this.logicalStorageGroupName = logicalStorageGroupName;
this.fileFlushPolicy = fileFlushPolicy;
- storageGroupSysDir = SystemFileFactory.INSTANCE.getFile(systemDir, virtualStorageGroupId);
+ storageGroupSysDir = SystemFileFactory.INSTANCE.getFile(systemDir, dataRegionId);
this.tsFileManager =
- new TsFileManager(
- logicalStorageGroupName, virtualStorageGroupId, storageGroupSysDir.getPath());
+ new TsFileManager(logicalStorageGroupName, dataRegionId, storageGroupSysDir.getPath());
if (storageGroupSysDir.mkdirs()) {
logger.info(
"Storage Group system Directory {} doesn't exist, create it",
@@ -424,11 +423,7 @@ public class VirtualStorageGroupProcessor {
// start trim task at last
walTrimScheduleTask =
IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
- ThreadName.WAL_TRIM.getName()
- + "-"
- + logicalStorageGroupName
- + "-"
- + virtualStorageGroupId);
+ ThreadName.WAL_TRIM.getName() + "-" + logicalStorageGroupName + "-" + dataRegionId);
walTrimScheduleTask.scheduleWithFixedDelay(
this::trimTask,
config.getWalPoolTrimIntervalInMS(),
@@ -494,10 +489,8 @@ public class VirtualStorageGroupProcessor {
// log only when log interval exceeds recovery log interval
if (lastLogTime + config.getRecoveryLogIntervalInMs() < System.currentTimeMillis()) {
logger.info(
- "The virtual storage group {}[{}] has recovered {}%, please wait a moment.",
- logicalStorageGroupName,
- virtualStorageGroupId,
- recoveredFilesNum * 1.0 / filesToRecoverNum);
+ "The data region {}[{}] has recovered {}%, please wait a moment.",
+ logicalStorageGroupName, dataRegionId, recoveredFilesNum * 1.0 / filesToRecoverNum);
lastLogTime = System.currentTimeMillis();
}
}
@@ -581,9 +574,7 @@ public class VirtualStorageGroupProcessor {
initCompaction();
logger.info(
- "The virtual storage group {}[{}] is recovered successfully",
- logicalStorageGroupName,
- virtualStorageGroupId);
+ "The data region {}[{}] is recovered successfully", logicalStorageGroupName, dataRegionId);
}
private void initCompaction() {
@@ -593,7 +584,7 @@ public class VirtualStorageGroupProcessor {
+ "-"
+ logicalStorageGroupName
+ "-"
- + virtualStorageGroupId);
+ + dataRegionId);
timedCompactionScheduleTask.scheduleWithFixedDelay(
this::executeCompaction,
COMPACTION_TASK_SUBMIT_DELAY,
@@ -603,7 +594,7 @@ public class VirtualStorageGroupProcessor {
private void recoverCompaction() throws Exception {
CompactionRecoverManager compactionRecoverManager =
- new CompactionRecoverManager(tsFileManager, logicalStorageGroupName, virtualStorageGroupId);
+ new CompactionRecoverManager(tsFileManager, logicalStorageGroupName, dataRegionId);
compactionRecoverManager.recoverInnerSpaceCompaction(true);
compactionRecoverManager.recoverInnerSpaceCompaction(false);
compactionRecoverManager.recoverCrossSpaceCompaction();
@@ -666,8 +657,7 @@ public class VirtualStorageGroupProcessor {
List<File> upgradeFiles = new ArrayList<>();
for (String baseDir : folders) {
File fileFolder =
- fsFactory.getFile(
- baseDir + File.separator + logicalStorageGroupName, virtualStorageGroupId);
+ fsFactory.getFile(baseDir + File.separator + logicalStorageGroupName, dataRegionId);
if (!fileFolder.exists()) {
continue;
}
@@ -746,10 +736,10 @@ public class VirtualStorageGroupProcessor {
if (fileTime > currentTime) {
throw new StorageGroupProcessorException(
String.format(
- "virtual storage group %s[%s] is down, because the time of tsfile %s is larger than system current time, "
+ "data region %s[%s] is down, because the time of tsfile %s is larger than system current time, "
+ "file time is %d while system current time is %d, please check it.",
logicalStorageGroupName,
- virtualStorageGroupId,
+ dataRegionId,
tsFile.getAbsolutePath(),
fileTime,
currentTime));
@@ -767,10 +757,7 @@ public class VirtualStorageGroupProcessor {
TsFileRecoverPerformer recoverPerformer =
new TsFileRecoverPerformer(
- logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId
- + FILE_NAME_SEPARATOR,
+ logicalStorageGroupName + File.separator + dataRegionId + FILE_NAME_SEPARATOR,
tsFileResource,
isSeq,
i == tsFiles.size() - 1,
@@ -805,7 +792,7 @@ public class VirtualStorageGroupProcessor {
if (isSeq) {
tsFileProcessor =
new TsFileProcessor(
- virtualStorageGroupId,
+ dataRegionId,
storageGroupInfo,
tsFileResource,
this::closeUnsealedTsFileProcessorCallBack,
@@ -821,7 +808,7 @@ public class VirtualStorageGroupProcessor {
} else {
tsFileProcessor =
new TsFileProcessor(
- virtualStorageGroupId,
+ dataRegionId,
storageGroupInfo,
tsFileResource,
this::closeUnsealedTsFileProcessorCallBack,
@@ -1476,7 +1463,7 @@ public class VirtualStorageGroupProcessor {
TsFileNameGenerator.generateNewTsFilePathWithMkdir(
sequence,
logicalStorageGroupName,
- virtualStorageGroupId,
+ dataRegionId,
timePartitionId,
System.currentTimeMillis(),
version,
@@ -1492,7 +1479,7 @@ public class VirtualStorageGroupProcessor {
if (sequence) {
tsFileProcessor =
new TsFileProcessor(
- logicalStorageGroupName + File.separator + virtualStorageGroupId,
+ logicalStorageGroupName + File.separator + dataRegionId,
fsFactory.getFileWithParent(filePath),
storageGroupInfo,
this::closeUnsealedTsFileProcessorCallBack,
@@ -1501,7 +1488,7 @@ public class VirtualStorageGroupProcessor {
} else {
tsFileProcessor =
new TsFileProcessor(
- logicalStorageGroupName + File.separator + virtualStorageGroupId,
+ logicalStorageGroupName + File.separator + dataRegionId,
fsFactory.getFileWithParent(filePath),
storageGroupInfo,
this::closeUnsealedTsFileProcessorCallBack,
@@ -1554,7 +1541,7 @@ public class VirtualStorageGroupProcessor {
if (System.currentTimeMillis() - startTime > 60_000) {
logger.warn(
"{} has spent {}s to wait for closing one tsfile.",
- logicalStorageGroupName + "-" + this.virtualStorageGroupId,
+ logicalStorageGroupName + "-" + this.dataRegionId,
(System.currentTimeMillis() - startTime) / 1000);
}
}
@@ -1563,7 +1550,7 @@ public class VirtualStorageGroupProcessor {
logger.error(
"syncCloseOneTsFileProcessor error occurs while waiting for closing the storage "
+ "group {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId,
+ logicalStorageGroupName + "-" + dataRegionId,
e);
}
}
@@ -1598,8 +1585,7 @@ public class VirtualStorageGroupProcessor {
timePartitionIdVersionControllerMap.remove(tsFileProcessor.getTimeRangeId());
}
logger.info(
- "close a sequence tsfile processor {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId);
+ "close a sequence tsfile processor {}", logicalStorageGroupName + "-" + dataRegionId);
} else {
closingUnSequenceTsFileProcessor.add(tsFileProcessor);
tsFileProcessor.asyncClose();
@@ -1621,12 +1607,11 @@ public class VirtualStorageGroupProcessor {
public void deleteFolder(String systemDir) {
logger.info(
"{} will close all files for deleting data folder {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId,
+ logicalStorageGroupName + "-" + dataRegionId,
systemDir);
writeLock("deleteFolder");
try {
- File storageGroupFolder =
- SystemFileFactory.INSTANCE.getFile(systemDir, virtualStorageGroupId);
+ File storageGroupFolder = SystemFileFactory.INSTANCE.getFile(systemDir, dataRegionId);
if (storageGroupFolder.exists()) {
org.apache.iotdb.db.utils.FileUtils.deleteDirectory(storageGroupFolder);
}
@@ -1667,7 +1652,7 @@ public class VirtualStorageGroupProcessor {
public void syncDeleteDataFiles() {
logger.info(
"{} will close all files for deleting data files",
- logicalStorageGroupName + "-" + virtualStorageGroupId);
+ logicalStorageGroupName + "-" + dataRegionId);
writeLock("syncDeleteDataFiles");
try {
@@ -1693,8 +1678,7 @@ public class VirtualStorageGroupProcessor {
private void deleteAllSGFolders(List<String> folder) {
for (String tsfilePath : folder) {
File storageGroupFolder =
- fsFactory.getFile(
- tsfilePath, logicalStorageGroupName + File.separator + virtualStorageGroupId);
+ fsFactory.getFile(tsfilePath, logicalStorageGroupName + File.separator + dataRegionId);
if (storageGroupFolder.exists()) {
org.apache.iotdb.db.utils.FileUtils.deleteDirectory(storageGroupFolder);
}
@@ -1705,14 +1689,13 @@ public class VirtualStorageGroupProcessor {
public synchronized void checkFilesTTL() {
if (dataTTL == Long.MAX_VALUE) {
logger.debug(
- "{}: TTL not set, ignore the check",
- logicalStorageGroupName + "-" + virtualStorageGroupId);
+ "{}: TTL not set, ignore the check", logicalStorageGroupName + "-" + dataRegionId);
return;
}
long ttlLowerBound = System.currentTimeMillis() - dataTTL;
logger.debug(
"{}: TTL removing files before {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId,
+ logicalStorageGroupName + "-" + dataRegionId,
new Date(ttlLowerBound));
// copy to avoid concurrent modification of deletion
@@ -1766,7 +1749,7 @@ public class VirtualStorageGroupProcessor {
"Exceed sequence memtable flush interval, so flush working memtable of time partition {} in storage group {}[{}]",
tsFileProcessor.getTimeRangeId(),
logicalStorageGroupName,
- virtualStorageGroupId);
+ dataRegionId);
fileFlushPolicy.apply(this, tsFileProcessor, tsFileProcessor.isSequence());
}
}
@@ -1789,7 +1772,7 @@ public class VirtualStorageGroupProcessor {
"Exceed unsequence memtable flush interval, so flush working memtable of time partition {} in storage group {}[{}]",
tsFileProcessor.getTimeRangeId(),
logicalStorageGroupName,
- virtualStorageGroupId);
+ dataRegionId);
fileFlushPolicy.apply(this, tsFileProcessor, tsFileProcessor.isSequence());
}
}
@@ -1813,7 +1796,7 @@ public class VirtualStorageGroupProcessor {
"Exceed tsfile close interval, so close TsFileProcessor of time partition {} in storage group {}[{}]",
tsFileProcessor.getTimeRangeId(),
logicalStorageGroupName,
- virtualStorageGroupId);
+ dataRegionId);
asyncCloseOneTsFileProcessor(true, tsFileProcessor);
}
}
@@ -1829,7 +1812,7 @@ public class VirtualStorageGroupProcessor {
"Exceed tsfile close interval, so close TsFileProcessor of time partition {} in storage group {}[{}]",
tsFileProcessor.getTimeRangeId(),
logicalStorageGroupName,
- virtualStorageGroupId);
+ dataRegionId);
asyncCloseOneTsFileProcessor(false, tsFileProcessor);
}
}
@@ -1850,7 +1833,7 @@ public class VirtualStorageGroupProcessor {
if (System.currentTimeMillis() - startTime > 60_000) {
logger.warn(
"{} has spent {}s to wait for closing all TsFiles.",
- logicalStorageGroupName + "-" + this.virtualStorageGroupId,
+ logicalStorageGroupName + "-" + this.dataRegionId,
(System.currentTimeMillis() - startTime) / 1000);
}
}
@@ -1858,7 +1841,7 @@ public class VirtualStorageGroupProcessor {
logger.error(
"CloseFileNodeCondition error occurs while waiting for closing the storage "
+ "group {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId,
+ logicalStorageGroupName + "-" + dataRegionId,
e);
Thread.currentThread().interrupt();
}
@@ -1871,7 +1854,7 @@ public class VirtualStorageGroupProcessor {
try {
logger.info(
"async force close all files in storage group: {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId);
+ logicalStorageGroupName + "-" + dataRegionId);
// to avoid concurrent modification problem, we need a new array list
for (TsFileProcessor tsFileProcessor :
new ArrayList<>(workSequenceTsFileProcessors.values())) {
@@ -1893,7 +1876,7 @@ public class VirtualStorageGroupProcessor {
try {
logger.info(
"force close all processors in storage group: {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId);
+ logicalStorageGroupName + "-" + dataRegionId);
// to avoid concurrent modification problem, we need a new array list
for (TsFileProcessor tsFileProcessor :
new ArrayList<>(workSequenceTsFileProcessors.values())) {
@@ -2358,7 +2341,7 @@ public class VirtualStorageGroupProcessor {
}
logger.info(
"signal closing storage group condition in {}",
- logicalStorageGroupName + "-" + virtualStorageGroupId);
+ logicalStorageGroupName + "-" + dataRegionId);
}
private void executeCompaction() {
@@ -2907,7 +2890,7 @@ public class VirtualStorageGroupProcessor {
DirectoryManager.getInstance().getNextFolderForUnSequenceFile(),
logicalStorageGroupName
+ File.separatorChar
- + virtualStorageGroupId
+ + dataRegionId
+ File.separatorChar
+ filePartitionId
+ File.separator
@@ -2929,7 +2912,7 @@ public class VirtualStorageGroupProcessor {
DirectoryManager.getInstance().getNextFolderForSequenceFile(),
logicalStorageGroupName
+ File.separatorChar
- + virtualStorageGroupId
+ + dataRegionId
+ File.separatorChar
+ filePartitionId
+ File.separator
@@ -3161,13 +3144,13 @@ public class VirtualStorageGroupProcessor {
return tsFileManager.getTsFileList(false);
}
- public String getVirtualStorageGroupId() {
- return virtualStorageGroupId;
+ public String getDataRegionId() {
+ return dataRegionId;
}
- /** @return virtual storage group path, like root.sg1/0 */
+ /** @return data region path, like root.sg1/0 */
public String getStorageGroupPath() {
- return logicalStorageGroupName + File.separator + virtualStorageGroupId;
+ return logicalStorageGroupName + File.separator + dataRegionId;
}
public StorageGroupInfo getStorageGroupInfo() {
@@ -3264,7 +3247,7 @@ public class VirtualStorageGroupProcessor {
tsFileManager.setAllowCompaction(false);
List<AbstractCompactionTask> runningTasks =
CompactionTaskManager.getInstance()
- .abortCompaction(logicalStorageGroupName + "-" + virtualStorageGroupId);
+ .abortCompaction(logicalStorageGroupName + "-" + dataRegionId);
while (CompactionTaskManager.getInstance().isAnyTaskInListStillRunning(runningTasks)) {
try {
TimeUnit.MILLISECONDS.sleep(10);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupInfo.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupInfo.java
index a8ba599515..e80a111cc9 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupInfo.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupInfo.java
@@ -34,7 +34,7 @@ import java.util.function.Supplier;
/** The storageGroupInfo records the total memory cost of the Storage Group. */
public class StorageGroupInfo {
- private VirtualStorageGroupProcessor virtualStorageGroupProcessor;
+ private DataRegion dataRegion;
/**
* The total Storage group memory cost, including unsealed TsFileResource, ChunkMetadata, WAL,
@@ -51,13 +51,13 @@ public class StorageGroupInfo {
/** A set of all unclosed TsFileProcessors in this SG */
private List<TsFileProcessor> reportedTsps = new CopyOnWriteArrayList<>();
- public StorageGroupInfo(VirtualStorageGroupProcessor virtualStorageGroupProcessor) {
- this.virtualStorageGroupProcessor = virtualStorageGroupProcessor;
+ public StorageGroupInfo(DataRegion dataRegion) {
+ this.dataRegion = dataRegion;
memoryCost = new AtomicLong();
}
- public VirtualStorageGroupProcessor getVirtualStorageGroupProcessor() {
- return virtualStorageGroupProcessor;
+ public DataRegion getVirtualStorageGroupProcessor() {
+ return dataRegion;
}
/** When create a new TsFileProcessor, call this method */
@@ -101,8 +101,8 @@ public class StorageGroupInfo {
}
public Supplier<ByteBuffer[]> getWalSupplier() {
- if (virtualStorageGroupProcessor != null) {
- return virtualStorageGroupProcessor::getWalDirectByteBuffer;
+ if (dataRegion != null) {
+ return dataRegion::getWalDirectByteBuffer;
} else { // only happens in test
return this::walSupplier;
}
@@ -119,8 +119,8 @@ public class StorageGroupInfo {
}
public Consumer<ByteBuffer[]> getWalConsumer() {
- if (virtualStorageGroupProcessor != null) {
- return virtualStorageGroupProcessor::releaseWalBuffer;
+ if (dataRegion != null) {
+ return dataRegion::releaseWalBuffer;
} else { // only happens in test
return this::walConsumer;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileManager.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileManager.java
index 8f03876d28..b816f28c42 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileManager.java
@@ -44,7 +44,7 @@ import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFF
public class TsFileManager {
private static final Logger LOGGER = LoggerFactory.getLogger(TsFileManager.class);
private String storageGroupName;
- private String virtualStorageGroup;
+ private String dataRegion;
private String storageGroupDir;
/** Serialize queries, delete resource files, compaction cleanup files */
@@ -60,11 +60,10 @@ public class TsFileManager {
private boolean allowCompaction = true;
- public TsFileManager(
- String storageGroupName, String virtualStorageGroup, String storageGroupDir) {
+ public TsFileManager(String storageGroupName, String dataRegion, String storageGroupDir) {
this.storageGroupName = storageGroupName;
this.storageGroupDir = storageGroupDir;
- this.virtualStorageGroup = virtualStorageGroup;
+ this.dataRegion = dataRegion;
}
public List<TsFileResource> getTsFileList(boolean sequence) {
@@ -355,12 +354,12 @@ public class TsFileManager {
this.allowCompaction = allowCompaction;
}
- public String getVirtualStorageGroup() {
- return virtualStorageGroup;
+ public String getDataRegion() {
+ return dataRegion;
}
- public void setVirtualStorageGroup(String virtualStorageGroup) {
- this.virtualStorageGroup = virtualStorageGroup;
+ public void setDataRegion(String dataRegion) {
+ this.dataRegion = dataRegion;
}
public List<TsFileResource> getSequenceRecoverTsFileResources() {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 5fd5aad7f5..ec7ef57821 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -35,7 +35,7 @@ import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.UpdateEndTimeCallBack;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.UpdateEndTimeCallBack;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.WriteProcessRejectException;
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 6c9e83c8b5..610e6c47e4 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
@@ -23,9 +23,9 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.SettleTsFileCallBack;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.UpgradeTsFileResourceCallBack;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator.TsFileName;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.SettleTsFileCallBack;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.UpgradeTsFileResourceCallBack;
import org.apache.iotdb.db.engine.storagegroup.timeindex.DeviceTimeIndex;
import org.apache.iotdb.db.engine.storagegroup.timeindex.FileTimeIndex;
import org.apache.iotdb.db.engine.storagegroup.timeindex.ITimeIndex;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/HashVirtualPartitioner.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/HashVirtualPartitioner.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/HashVirtualPartitioner.java
rename to server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/HashVirtualPartitioner.java
index 14537e8ef7..088f6df8d9 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/HashVirtualPartitioner.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/HashVirtualPartitioner.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.engine.storagegroup.virtualSg;
+package org.apache.iotdb.db.engine.storagegroup.dataregion;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.metadata.path.PartialPath;
@@ -25,7 +25,7 @@ public class HashVirtualPartitioner implements VirtualPartitioner {
/** total number of virtual storage groups */
public static int STORAGE_GROUP_NUM =
- IoTDBDescriptor.getInstance().getConfig().getVirtualStorageGroupNum();
+ IoTDBDescriptor.getInstance().getConfig().getDataRegionNum();
private HashVirtualPartitioner() {}
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/dataregion/StorageGroupManager.java
similarity index 69%
rename from server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/StorageGroupManager.java
rename to server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/StorageGroupManager.java
index 3f895d07d8..2bcd4efa5a 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/dataregion/StorageGroupManager.java
@@ -16,14 +16,14 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.engine.storagegroup.virtualSg;
+package org.apache.iotdb.db.engine.storagegroup.dataregion;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.StorageGroupNotReadyException;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
@@ -59,7 +59,7 @@ public class StorageGroupManager {
VirtualPartitioner partitioner = HashVirtualPartitioner.getInstance();
/** all virtual storage group processor */
- VirtualStorageGroupProcessor[] virtualStorageGroupProcessor;
+ DataRegion[] dataRegion;
/**
* recover status of each virtual storage group processor, null if this logical storage group is
@@ -80,8 +80,7 @@ public class StorageGroupManager {
}
public StorageGroupManager(boolean needRecovering) {
- virtualStorageGroupProcessor =
- new VirtualStorageGroupProcessor[partitioner.getPartitionCount()];
+ dataRegion = new DataRegion[partitioner.getPartitionCount()];
isVsgReady = new AtomicBoolean[partitioner.getPartitionCount()];
boolean recoverReady = !needRecovering;
for (int i = 0; i < partitioner.getPartitionCount(); i++) {
@@ -91,60 +90,54 @@ public class StorageGroupManager {
/** push forceCloseAllWorkingTsFileProcessors down to all sg */
public void forceCloseAllWorkingTsFileProcessors() throws TsFileProcessorException {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.forceCloseAllWorkingTsFileProcessors();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.forceCloseAllWorkingTsFileProcessors();
}
}
}
/** push syncCloseAllWorkingTsFileProcessors down to all sg */
public void syncCloseAllWorkingTsFileProcessors() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
}
}
/** push check ttl down to all sg */
public void checkTTL() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.checkFilesTTL();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.checkFilesTTL();
}
}
}
/** push check sequence memtable flush interval down to all sg */
public void timedFlushSeqMemTable() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.timedFlushSeqMemTable();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.timedFlushSeqMemTable();
}
}
}
/** push check unsequence memtable flush interval down to all sg */
public void timedFlushUnseqMemTable() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.timedFlushUnseqMemTable();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.timedFlushUnseqMemTable();
}
}
}
/** push check TsFileProcessor close interval down to all sg */
public void timedCloseTsFileProcessor() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.timedCloseTsFileProcessor();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.timedCloseTsFileProcessor();
}
}
}
@@ -157,23 +150,22 @@ public class StorageGroupManager {
*/
@SuppressWarnings("java:S2445")
// actually storageGroupMNode is a unique object on the mtree, synchronize it is reasonable
- public VirtualStorageGroupProcessor getProcessor(
- PartialPath partialPath, IStorageGroupMNode storageGroupMNode)
+ public DataRegion getProcessor(PartialPath partialPath, IStorageGroupMNode storageGroupMNode)
throws StorageGroupProcessorException, StorageEngineException {
int loc = partitioner.deviceToVirtualStorageGroupId(partialPath);
- VirtualStorageGroupProcessor processor = virtualStorageGroupProcessor[loc];
+ DataRegion processor = dataRegion[loc];
if (processor == null) {
// if finish recover
if (isVsgReady[loc].get()) {
synchronized (storageGroupMNode) {
- processor = virtualStorageGroupProcessor[loc];
+ processor = dataRegion[loc];
if (processor == null) {
processor =
StorageEngine.getInstance()
.buildNewStorageGroupProcessor(
storageGroupMNode.getPartialPath(), storageGroupMNode, String.valueOf(loc));
- virtualStorageGroupProcessor[loc] = processor;
+ dataRegion[loc] = processor;
}
}
} else {
@@ -201,7 +193,7 @@ public class StorageGroupManager {
Callable<Void> recoverVsgTask =
() -> {
isVsgReady[cur].set(false);
- VirtualStorageGroupProcessor processor = null;
+ DataRegion processor = null;
try {
processor =
StorageEngine.getInstance()
@@ -217,7 +209,7 @@ public class StorageGroupManager {
e);
}
- virtualStorageGroupProcessor[cur] = processor;
+ dataRegion[cur] = processor;
isVsgReady[cur].set(true);
logger.info(
"Storage Group {} has been recovered {}/{}",
@@ -244,7 +236,7 @@ public class StorageGroupManager {
/** push closeStorageGroupProcessor operation down to all virtual storage group processors */
public void closeStorageGroupProcessor(boolean isSeq, boolean isSync) {
- for (VirtualStorageGroupProcessor processor : virtualStorageGroupProcessor) {
+ for (DataRegion processor : dataRegion) {
if (processor == null) {
continue;
}
@@ -253,7 +245,7 @@ public class StorageGroupManager {
logger.info(
"{} closing sg processor is called for closing {}, seq = {}",
isSync ? "sync" : "async",
- processor.getVirtualStorageGroupId() + "-" + processor.getLogicalStorageGroupName(),
+ processor.getDataRegionId() + "-" + processor.getLogicalStorageGroupName(),
isSeq);
}
@@ -288,11 +280,11 @@ public class StorageGroupManager {
/** push closeStorageGroupProcessor operation down to all virtual storage group processors */
public void closeStorageGroupProcessor(long partitionId, boolean isSeq, boolean isSync) {
- for (VirtualStorageGroupProcessor processor : virtualStorageGroupProcessor) {
+ for (DataRegion processor : dataRegion) {
if (processor != null) {
logger.info(
"async closing sg processor is called for closing {}, seq = {}, partitionId = {}",
- processor.getVirtualStorageGroupId() + "-" + processor.getLogicalStorageGroupName(),
+ processor.getDataRegionId() + "-" + processor.getLogicalStorageGroupName(),
isSeq,
partitionId);
processor.writeLock("VirtualCloseStorageGroupProcessor-242");
@@ -327,11 +319,9 @@ public class StorageGroupManager {
long planIndex,
TimePartitionFilter timePartitionFilter)
throws IOException {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.delete(
- path, startTime, endTime, planIndex, timePartitionFilter);
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.delete(path, startTime, endTime, planIndex, timePartitionFilter);
}
}
}
@@ -339,10 +329,9 @@ public class StorageGroupManager {
/** push countUpgradeFiles operation down to all virtual storage group processors */
public int countUpgradeFiles() {
int totalUpgradeFileNum = 0;
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- totalUpgradeFileNum += virtualStorageGroupProcessor.countUpgradeFiles();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ totalUpgradeFileNum += dataRegion.countUpgradeFiles();
}
}
@@ -351,10 +340,9 @@ public class StorageGroupManager {
/** push upgradeAll operation down to all virtual storage group processors */
public void upgradeAll() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.upgrade();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.upgrade();
}
}
}
@@ -363,10 +351,9 @@ public class StorageGroupManager {
List<TsFileResource> seqResourcesToBeSettled,
List<TsFileResource> unseqResourcesToBeSettled,
List<String> tsFilePaths) {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.addSettleFilesToList(
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.addSettleFilesToList(
seqResourcesToBeSettled, unseqResourcesToBeSettled, tsFilePaths);
}
}
@@ -374,37 +361,34 @@ public class StorageGroupManager {
/** push mergeAll operation down to all virtual storage group processors */
public void mergeAll() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.compact();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.compact();
}
}
}
/** push syncDeleteDataFiles operation down to all virtual storage group processors */
public void syncDeleteDataFiles() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.syncDeleteDataFiles();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.syncDeleteDataFiles();
}
}
}
/** push setTTL operation down to all virtual storage group processors */
public void setTTL(long dataTTL) {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.setDataTTL(dataTTL);
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.setDataTTL(dataTTL);
}
}
}
/** push deleteStorageGroup operation down to all virtual storage group processors */
public void deleteStorageGroupSystemFolder(String path) {
- for (VirtualStorageGroupProcessor processor : virtualStorageGroupProcessor) {
+ for (DataRegion processor : dataRegion) {
if (processor != null) {
processor.deleteFolder(path);
}
@@ -414,11 +398,10 @@ public class StorageGroupManager {
/** push getAllClosedStorageGroupTsFile operation down to all virtual storage group processors */
public void getAllClosedStorageGroupTsFile(
PartialPath storageGroupName, Map<PartialPath, Map<Long, List<TsFileResource>>> ret) {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- List<TsFileResource> allResources = virtualStorageGroupProcessor.getSequenceFileList();
- allResources.addAll(virtualStorageGroupProcessor.getUnSequenceFileList());
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ List<TsFileResource> allResources = dataRegion.getSequenceFileList();
+ allResources.addAll(dataRegion.getUnSequenceFileList());
for (TsFileResource tsfile : allResources) {
if (!tsfile.isClosed()) {
continue;
@@ -434,20 +417,18 @@ public class StorageGroupManager {
/** push setPartitionVersionToMax operation down to all virtual storage group processors */
public void setPartitionVersionToMax(long partitionId, long newMaxVersion) {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.setPartitionFileVersionToMax(partitionId, newMaxVersion);
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.setPartitionFileVersionToMax(partitionId, newMaxVersion);
}
}
}
/** push removePartitions operation down to all virtual storage group processors */
public void removePartitions(TimePartitionFilter filter) {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.removePartitions(filter);
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.removePartitions(filter);
}
}
}
@@ -457,18 +438,15 @@ public class StorageGroupManager {
*/
public void getWorkingStorageGroupPartitions(
String storageGroupName, Map<String, List<Pair<Long, Boolean>>> res) {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
List<Pair<Long, Boolean>> partitionIdList = new ArrayList<>();
- for (TsFileProcessor tsFileProcessor :
- virtualStorageGroupProcessor.getWorkSequenceTsFileProcessors()) {
+ for (TsFileProcessor tsFileProcessor : dataRegion.getWorkSequenceTsFileProcessors()) {
Pair<Long, Boolean> tmpPair = new Pair<>(tsFileProcessor.getTimeRangeId(), true);
partitionIdList.add(tmpPair);
}
- for (TsFileProcessor tsFileProcessor :
- virtualStorageGroupProcessor.getWorkUnsequenceTsFileProcessors()) {
+ for (TsFileProcessor tsFileProcessor : dataRegion.getWorkUnsequenceTsFileProcessors()) {
Pair<Long, Boolean> tmpPair = new Pair<>(tsFileProcessor.getTimeRangeId(), false);
partitionIdList.add(tmpPair);
}
@@ -480,21 +458,20 @@ public class StorageGroupManager {
/** release resource of direct wal buffer */
public void releaseWalDirectByteBufferPool() {
- for (VirtualStorageGroupProcessor virtualStorageGroupProcessor :
- this.virtualStorageGroupProcessor) {
- if (virtualStorageGroupProcessor != null) {
- virtualStorageGroupProcessor.releaseWalDirectByteBufferPool();
+ for (DataRegion dataRegion : this.dataRegion) {
+ if (dataRegion != null) {
+ dataRegion.releaseWalDirectByteBufferPool();
}
}
}
/** only for test */
public void reset() {
- Arrays.fill(virtualStorageGroupProcessor, null);
+ Arrays.fill(dataRegion, null);
}
public void stopSchedulerPool() {
- for (VirtualStorageGroupProcessor vsg : this.virtualStorageGroupProcessor) {
+ for (DataRegion vsg : this.dataRegion) {
if (vsg != null) {
ThreadUtils.stopThreadPool(
vsg.getTimedCompactionScheduleTask(), ThreadName.COMPACTION_SCHEDULE);
@@ -508,13 +485,13 @@ public class StorageGroupManager {
}
public void setAllowCompaction(boolean allowCompaction) {
- for (VirtualStorageGroupProcessor processor : virtualStorageGroupProcessor) {
+ for (DataRegion processor : dataRegion) {
processor.setAllowCompaction(allowCompaction);
}
}
public void abortCompaction() {
- for (VirtualStorageGroupProcessor processor : virtualStorageGroupProcessor) {
+ for (DataRegion processor : dataRegion) {
processor.abortCompaction();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualPartitioner.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/VirtualPartitioner.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualPartitioner.java
rename to server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/VirtualPartitioner.java
index ea6ab05d20..18c3dc5b61 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/VirtualPartitioner.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/dataregion/VirtualPartitioner.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.engine.storagegroup.virtualSg;
+package org.apache.iotdb.db.engine.storagegroup.dataregion;
import org.apache.iotdb.db.metadata.path.PartialPath;
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/tag/TagManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/tag/TagManager.java
index b883644a0f..97f1749731 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/tag/TagManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/tag/TagManager.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.metadata.tag;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -138,11 +138,9 @@ public class TagManager {
// if ordered by heat, we sort all the timeseries by the descending order of the last insert
// timestamp
if (plan.isOrderByHeat()) {
- List<VirtualStorageGroupProcessor> list;
+ List<DataRegion> list;
try {
- Pair<
- List<VirtualStorageGroupProcessor>,
- Map<VirtualStorageGroupProcessor, List<PartialPath>>>
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>>
lockListAndProcessorToSeriesMapPair =
StorageEngine.getInstance()
.mergeLock(
@@ -150,7 +148,7 @@ public class TagManager {
.map(IMeasurementMNode::getMeasurementPath)
.collect(toList()));
list = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java
index 5a1168fac3..f44d1e9887 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriver.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.db.mpp.execution;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.idtable.IDTable;
import org.apache.iotdb.db.metadata.path.PartialPath;
@@ -197,7 +197,7 @@ public class DataDriver implements Driver {
* QueryDataSource needed for this query
*/
public QueryDataSource initQueryDataSourceCache() throws QueryProcessException {
- VirtualStorageGroupProcessor dataRegion = driverContext.getDataRegion();
+ DataRegion dataRegion = driverContext.getDataRegion();
dataRegion.readLock();
try {
List<PartialPath> pathList =
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriverContext.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriverContext.java
index 52113e5586..43a9c0ab1c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriverContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/DataDriverContext.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.mpp.execution;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.mpp.operator.source.SourceOperator;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
@@ -28,14 +28,14 @@ import java.util.List;
public class DataDriverContext extends DriverContext {
private final List<PartialPath> paths;
private final Filter timeFilter;
- private final VirtualStorageGroupProcessor dataRegion;
+ private final DataRegion dataRegion;
private final List<SourceOperator> sourceOperators;
public DataDriverContext(
FragmentInstanceContext fragmentInstanceContext,
List<PartialPath> paths,
Filter timeFilter,
- VirtualStorageGroupProcessor dataRegion,
+ DataRegion dataRegion,
List<SourceOperator> sourceOperators) {
super(fragmentInstanceContext);
this.paths = paths;
@@ -52,7 +52,7 @@ public class DataDriverContext extends DriverContext {
return timeFilter;
}
- public VirtualStorageGroupProcessor getDataRegion() {
+ public DataRegion getDataRegion() {
return dataRegion;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java
index 6ba68e55a7..ca17860d4b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/FragmentInstanceManager.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.mpp.execution;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.metadata.schemaregion.SchemaRegion;
import org.apache.iotdb.db.mpp.common.FragmentInstanceId;
import org.apache.iotdb.db.mpp.schedule.FragmentInstanceScheduler;
@@ -78,7 +78,7 @@ public class FragmentInstanceManager {
}
public FragmentInstanceInfo execDataQueryFragmentInstance(
- FragmentInstance instance, VirtualStorageGroupProcessor dataRegion) {
+ FragmentInstance instance, DataRegion dataRegion) {
FragmentInstanceId instanceId = instance.getId();
FragmentInstanceExecution execution =
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
index 1dbc265c0c..f9130534f4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/sql/planner/LocalExecutionPlanner.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.mpp.sql.planner;
import org.apache.iotdb.commons.cluster.Endpoint;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.metadata.schemaregion.SchemaRegion;
import org.apache.iotdb.db.mpp.buffer.DataBlockManager;
@@ -84,7 +84,7 @@ public class LocalExecutionPlanner {
PlanNode plan,
FragmentInstanceContext instanceContext,
Filter timeFilter,
- VirtualStorageGroupProcessor dataRegion) {
+ DataRegion dataRegion) {
LocalExecutionPlanContext context = new LocalExecutionPlanContext(instanceContext);
Operator root = plan.accept(new Visitor(), context);
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
index c0476c794e..f47b86c295 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.qp.executor;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
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 eb6a47f4f0..904b880815 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
@@ -34,9 +34,9 @@ import org.apache.iotdb.db.engine.cache.ChunkCache;
import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.engine.cq.ContinuousQueryService;
import org.apache.iotdb.db.engine.flush.pool.FlushTaskPoolManager;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.engine.trigger.service.TriggerRegistrationService;
import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.ContinuousQueryException;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
index 0a6cae9b6b..2eba63718f 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/DeletePlan.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.qp.physical.crud;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.qp.logical.Operator;
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java
index a6e82372e9..fab4108e54 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/DeleteTimeSeriesPlan.java
@@ -18,7 +18,7 @@
*/
package org.apache.iotdb.db.qp.physical.sys;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor.TimePartitionFilter;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion.TimePartitionFilter;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.qp.logical.Operator;
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
index a5ecb90dbc..8a19f55b2f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.query.control;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.idtable.IDTable;
@@ -122,13 +122,12 @@ public class QueryResourceManager {
* under the virtual storage group
*/
public void initQueryDataSourceCache(
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap,
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap,
QueryContext context,
Filter timeFilter)
throws QueryProcessException {
- for (Map.Entry<VirtualStorageGroupProcessor, List<PartialPath>> entry :
- processorToSeriesMap.entrySet()) {
- VirtualStorageGroupProcessor processor = entry.getKey();
+ for (Map.Entry<DataRegion, List<PartialPath>> entry : processorToSeriesMap.entrySet()) {
+ DataRegion processor = entry.getKey();
List<PartialPath> pathList =
entry.getValue().stream().map(IDTable::translateQueryPath).collect(Collectors.toList());
@@ -172,8 +171,7 @@ public class QueryResourceManager {
cachedQueryDataSource = cachedQueryDataSourcesMap.get(queryId).get(storageGroupPath);
} else {
// QueryDataSource is never cached in cluster mode
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath());
+ DataRegion processor = StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath());
PartialPath translatedPath = IDTable.translateQueryPath(selectedPath);
cachedQueryDataSource =
processor.query(
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
index 8b0409ea21..6034f22540 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.AlignedPath;
@@ -120,11 +120,10 @@ public class GroupByWithValueFilterDataSet extends GroupByTimeEngineDataSet {
groupedPathList.addAll(pathToAggrIndexesMap.keySet());
groupedPathList.addAll(alignedPathToAggrIndexesMap.keySet());
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance().mergeLock(groupedPathList);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(groupedPathList);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index eccb268c6e..1292e2c90f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.query.dataset.groupby;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.AlignedPath;
@@ -114,11 +114,10 @@ public class GroupByWithoutValueFilterDataSet extends GroupByTimeEngineDataSet {
groupedPathList.addAll(pathToAggrIndexesMap.keySet());
groupedPathList.addAll(alignedPathToAggrIndexesMap.keySet());
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance().mergeLock(groupedPathList);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(groupedPathList);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
index 0ce0b25740..3840ae4325 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
@@ -23,7 +23,7 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.AlignedPath;
@@ -126,11 +126,10 @@ public class AggregationExecutor {
groupedPathList.addAll(alignedPathToAggrIndexesMap.keySet());
// TODO-Cluster: group the paths by storage group to reduce communications
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance().mergeLock(groupedPathList);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(groupedPathList);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
@@ -635,11 +634,10 @@ public class AggregationExecutor {
groupedPathList.addAll(pathToAggrIndexesMap.keySet());
groupedPathList.addAll(alignedPathToAggrIndexesMap.keySet());
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance().mergeLock(groupedPathList);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(groupedPathList);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
index e2c3e60038..fc332294da 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.query.executor;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.PartialPath;
@@ -91,10 +91,10 @@ public class FillQueryExecutor {
Filter timeFilter = initFillExecutorsAndContructTimeFilter(context);
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair = StorageEngine.getInstance().mergeLock(selectedSeries);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(selectedSeries);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
index c5dd1e25dd..d3501141ce 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.query.executor;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -260,10 +260,10 @@ public class LastQueryExecutor {
Map<String, Set<String>> deviceMeasurementsMap)
throws StorageEngineException, QueryProcessException, IOException {
// Acquire query resources for the rest series paths
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair = StorageEngine.getInstance().mergeLock(seriesPaths);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(seriesPaths);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
index 984234f6be..12994f159c 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.query.executor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.AlignedPath;
@@ -106,11 +106,10 @@ public class RawDataQueryExecutor {
}
List<ManagedSeriesReader> readersOfSelectedSeries = new ArrayList<>();
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance().mergeLock(queryPlan.getDeduplicatedPaths());
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(queryPlan.getDeduplicatedPaths());
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
@@ -249,11 +248,10 @@ public class RawDataQueryExecutor {
cached.subList(pathList.size(), previousSize).clear();
}
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance().mergeLock(notCachedPathList);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(notCachedPathList);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
index 8a52b5b7d5..42ff873f97 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.query.timegenerator;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.path.MeasurementPath;
@@ -85,10 +85,10 @@ public class ServerTimeGenerator extends TimeGenerator {
List<PartialPath> pathList = new ArrayList<>();
timeFilter = getPathListAndConstructTimeFilterFromExpression(expression, pathList);
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair = StorageEngine.getInstance().mergeLock(pathList);
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance().mergeLock(pathList);
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/ISyncFileManager.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/ISyncFileManager.java
index ff558c5ac4..a615bb336a 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/ISyncFileManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/ISyncFileManager.java
@@ -57,7 +57,7 @@ public interface ISyncFileManager {
/*
* the following 4 maps share same map structure
- * logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ * logicalSg -> <dataregion, <timeRangeId, tsfiles>>
*/
Map<String, Map<Long, Map<Long, Set<File>>>> getCurrentSealedLocalFilesMap();
@@ -67,6 +67,6 @@ public interface ISyncFileManager {
Map<String, Map<Long, Map<Long, Set<File>>>> getToBeSyncedFilesMap();
- // logicalSg -> <virtualSg, Set<timeRangeId>>
+ // logicalSg -> <dataregion, Set<timeRangeId>>
Map<String, Map<Long, Set<Long>>> getAllSGs();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java
index 6aa4c90396..54066198d8 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/manage/SyncFileManager.java
@@ -47,14 +47,14 @@ public class SyncFileManager implements ISyncFileManager {
private static final Logger LOGGER = LoggerFactory.getLogger(SyncFileManager.class);
/**
- * All storage groups on the disk where the current sync task is executed logicalSg -> <virtualSg,
- * timeRangeId>
+ * All storage groups on the disk where the current sync task is executed logicalSg ->
+ * <dataregion, timeRangeId>
*/
private Map<String, Map<Long, Set<Long>>> allSGs;
/**
* Key is storage group, value is all sealed tsfiles in the storage group. Inner key is time range
- * id, inner value is the set of current sealed tsfiles. logicalSg -> <virtualSg, <timeRangeId,
+ * id, inner value is the set of current sealed tsfiles. logicalSg -> <dataregion, <timeRangeId,
* tsfiles>>
*/
private Map<String, Map<Long, Map<Long, Set<File>>>> currentSealedLocalFilesMap;
@@ -62,21 +62,21 @@ public class SyncFileManager implements ISyncFileManager {
/**
* Key is storage group, value is all last local tsfiles in the storage group, which doesn't
* contains those tsfiles which are not synced successfully. Inner key is time range id, inner
- * value is the set of last local tsfiles. logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ * value is the set of last local tsfiles. logicalSg -> <dataregion, <timeRangeId, tsfiles>>
*/
private Map<String, Map<Long, Map<Long, Set<File>>>> lastLocalFilesMap;
/**
* Key is storage group, value is all deleted tsfiles which need to be synced to receiver end in
* the storage group. Inner key is time range id, inner value is the valid set of sealed tsfiles.
- * logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ * logicalSg -> <dataregion, <timeRangeId, tsfiles>>
*/
private Map<String, Map<Long, Map<Long, Set<File>>>> deletedFilesMap;
/**
* Key is storage group, value is all new tsfiles which need to be synced to receiver end in the
* storage group. Inner key is time range id, inner value is the valid set of new tsfiles.
- * logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ * logicalSg -> <dataregion, <timeRangeId, tsfiles>>
*/
private Map<String, Map<Long, Map<Long, Set<File>>>> toBeSyncedFilesMap;
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/SyncClient.java b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/SyncClient.java
index c15572592a..b9e3222cb3 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/SyncClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/sender/transfer/SyncClient.java
@@ -106,16 +106,16 @@ public class SyncClient implements ISyncClient {
private SyncService.Client serviceClient;
- // logicalSg -> <virtualSg, timeRangeId>
+ // logicalSg -> <dataregion, timeRangeId>
private Map<String, Map<Long, Set<Long>>> allSG;
- // logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ // logicalSg -> <dataregion, <timeRangeId, tsfiles>>
private Map<String, Map<Long, Map<Long, Set<File>>>> toBeSyncedFilesMap;
- // logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ // logicalSg -> <dataregion, <timeRangeId, tsfiles>>
private Map<String, Map<Long, Map<Long, Set<File>>>> deletedFilesMap;
- // logicalSg -> <virtualSg, <timeRangeId, tsfiles>>
+ // logicalSg -> <dataregion, <timeRangeId, tsfiles>>
private Map<String, Map<Long, Map<Long, Set<File>>>> lastLocalFilesMap;
/** If true, sync is in execution. */
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/virtualsg/DeviceMappingViewer.java b/server/src/main/java/org/apache/iotdb/db/tools/virtualsg/DeviceMappingViewer.java
index 31f632ab7c..c6b97e9d0e 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/virtualsg/DeviceMappingViewer.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/virtualsg/DeviceMappingViewer.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.tools.virtualsg;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.storagegroup.virtualSg.HashVirtualPartitioner;
+import org.apache.iotdb.db.engine.storagegroup.dataregion.HashVirtualPartitioner;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.LocalSchemaProcessor;
import org.apache.iotdb.db.metadata.path.PartialPath;
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
index a75fa9ecb8..43d9eddbec 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
@@ -26,8 +26,8 @@ import org.apache.iotdb.db.engine.memtable.IWritableMemChunk;
import org.apache.iotdb.db.engine.memtable.IWritableMemChunkGroup;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.metadata.DataTypeMismatchException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -96,8 +96,7 @@ public class LogReplayer {
* finds the logNode of the TsFile given by insertFilePath and logNodePrefix, reads the WALs from
* the logNode and redoes them into a given MemTable and ModificationFile.
*/
- public void replayLogs(
- Supplier<ByteBuffer[]> supplier, VirtualStorageGroupProcessor virtualStorageGroupProcessor) {
+ public void replayLogs(Supplier<ByteBuffer[]> supplier, DataRegion dataRegion) {
WriteLogNode logNode =
MultiFileLogNodeManager.getInstance()
.getNode(
@@ -110,7 +109,7 @@ public class LogReplayer {
try {
PhysicalPlan plan = logReader.next();
if (plan instanceof InsertPlan) {
- replayInsert((InsertPlan) plan, virtualStorageGroupProcessor);
+ replayInsert((InsertPlan) plan, dataRegion);
} else if (plan instanceof DeletePlan) {
replayDelete((DeletePlan) plan);
}
@@ -160,8 +159,7 @@ public class LogReplayer {
}
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
- private void replayInsert(
- InsertPlan plan, VirtualStorageGroupProcessor virtualStorageGroupProcessor)
+ private void replayInsert(InsertPlan plan, DataRegion dataRegion)
throws WriteProcessException, QueryProcessException {
if (currentTsFileResource != null) {
long minTime, maxTime;
@@ -194,7 +192,7 @@ public class LogReplayer {
plan.setMeasurementMNodes(new IMeasurementMNode[plan.getMeasurements().length]);
try {
if (IoTDBDescriptor.getInstance().getConfig().isEnableIDTable()) {
- virtualStorageGroupProcessor.getIdTable().getSeriesSchemas(plan);
+ dataRegion.getIdTable().getSeriesSchemas(plan);
} else {
IoTDB.schemaProcessor.getSeriesSchemasAndReadLockDevice(plan);
plan.setDeviceID(DeviceIDFactory.getInstance().getDeviceID(plan.getDevicePath()));
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
index 730aa75d27..30157401d3 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java
@@ -26,8 +26,8 @@ import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.utils.FileLoaderUtils;
import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
@@ -68,7 +68,7 @@ public class TsFileRecoverPerformer {
private final String logNodePrefix;
private final TsFileResource tsFileResource;
private final boolean sequence;
- private VirtualStorageGroupProcessor virtualStorageGroupProcessor;
+ private DataRegion dataRegion;
/** @param isLastFile whether this TsFile is the last file of its partition */
public TsFileRecoverPerformer(
@@ -76,12 +76,12 @@ public class TsFileRecoverPerformer {
TsFileResource currentTsFileResource,
boolean sequence,
boolean isLastFile,
- VirtualStorageGroupProcessor storageGroupProcessor) {
+ DataRegion storageGroupProcessor) {
this.filePath = currentTsFileResource.getTsFilePath();
this.logNodePrefix = logNodePrefix;
this.tsFileResource = currentTsFileResource;
this.sequence = sequence;
- this.virtualStorageGroupProcessor = storageGroupProcessor;
+ this.dataRegion = storageGroupProcessor;
}
/**
@@ -287,7 +287,7 @@ public class TsFileRecoverPerformer {
tsFileResource,
recoverMemTable,
sequence);
- logReplayer.replayLogs(supplier, virtualStorageGroupProcessor);
+ logReplayer.replayLogs(supplier, dataRegion);
try {
if (!recoverMemTable.isEmpty()) {
// flush logs
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifierUT.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifierUT.java
index 5824bf5790..0a18c1d21b 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifierUT.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifierUT.java
@@ -51,7 +51,7 @@ public class TsFileIdentifierUT {
Assert.assertEquals(firstInfo.getFilename(), "1-1-0-0.tsfile");
Assert.assertEquals(firstInfo.getLogicalStorageGroupName(), "root.test.sg");
Assert.assertEquals(firstInfo.getTimePartitionId(), "0");
- Assert.assertEquals(firstInfo.getVirtualStorageGroupId(), "0");
+ Assert.assertEquals(firstInfo.getDataRegionId(), "0");
Assert.assertTrue(firstInfo.isSequence());
String secondPath =
@@ -69,7 +69,7 @@ public class TsFileIdentifierUT {
Assert.assertEquals(secondInfo.getFilename(), "999-3-24-12.tsfile");
Assert.assertEquals(secondInfo.getLogicalStorageGroupName(), "root.test.sg");
Assert.assertEquals(secondInfo.getTimePartitionId(), "426");
- Assert.assertEquals(secondInfo.getVirtualStorageGroupId(), "0");
+ Assert.assertEquals(secondInfo.getDataRegionId(), "0");
Assert.assertFalse(secondInfo.isSequence());
String illegalPath =
@@ -95,7 +95,7 @@ public class TsFileIdentifierUT {
TsFileIdentifier firstInfo = TsFileIdentifier.getFileIdentifierFromInfoString(firstInfoString);
Assert.assertEquals(firstInfo.getFilename(), "1-1-0-0.tsfile");
Assert.assertEquals(firstInfo.getTimePartitionId(), "0");
- Assert.assertEquals(firstInfo.getVirtualStorageGroupId(), "0");
+ Assert.assertEquals(firstInfo.getDataRegionId(), "0");
Assert.assertEquals(firstInfo.getLogicalStorageGroupName(), "root.test.sg");
Assert.assertTrue(firstInfo.isSequence());
@@ -106,7 +106,7 @@ public class TsFileIdentifierUT {
TsFileIdentifier.getFileIdentifierFromInfoString(secondInfoString);
Assert.assertEquals(secondInfo.getFilename(), "666-888-222-131.tsfile");
Assert.assertEquals(secondInfo.getTimePartitionId(), "425");
- Assert.assertEquals(secondInfo.getVirtualStorageGroupId(), "0");
+ Assert.assertEquals(secondInfo.getDataRegionId(), "0");
Assert.assertEquals(secondInfo.getLogicalStorageGroupName(), "root.test.sg");
Assert.assertFalse(secondInfo.isSequence());
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionTest.java
index 3e8dfa8bb3..cc66af6cd1 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionTest.java
@@ -23,10 +23,10 @@ import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.task.RewriteCrossSpaceCompactionTask;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -577,8 +577,8 @@ public class RewriteCrossSpaceCompactionTest extends AbstractCompactionTest {
*/
@Test
public void testOneDeletionDuringCompaction() throws Exception {
- VirtualStorageGroupProcessor vsgp =
- new VirtualStorageGroupProcessor(
+ DataRegion vsgp =
+ new DataRegion(
STORAGE_GROUP_DIR.getPath(),
"0",
new TsFileFlushPolicy.DirectFlushPolicy(),
@@ -687,8 +687,8 @@ public class RewriteCrossSpaceCompactionTest extends AbstractCompactionTest {
*/
@Test
public void testSeveralDeletionsDuringCompaction() throws Exception {
- VirtualStorageGroupProcessor vsgp =
- new VirtualStorageGroupProcessor(
+ DataRegion vsgp =
+ new DataRegion(
STORAGE_GROUP_DIR.getPath(),
"0",
new TsFileFlushPolicy.DirectFlushPolicy(),
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java
index d1744ceea0..114ec93a6e 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java
@@ -28,10 +28,10 @@ import org.apache.iotdb.db.engine.compaction.utils.CompactionConfigRestorer;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.path.PartialPath;
@@ -186,7 +186,7 @@ public class SizeTieredCompactionRecoverTest {
File timePartitionDir = new File(SEQ_FILE_DIR);
File f = new File(timePartitionDir.getParent() + File.separator + "test.tmp");
f.createNewFile();
- new VirtualStorageGroupProcessor(
+ new DataRegion(
TestConstant.BASE_OUTPUT_PATH + File.separator + "data" + File.separator + "sequence",
"0",
new TsFileFlushPolicy.DirectFlushPolicy(),
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
index 5c273ba60f..f367c63780 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
@@ -25,8 +25,8 @@ import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -170,12 +170,11 @@ public class DeletionFileNodeTest {
processorName + TsFileConstant.PATH_SEPARATOR + measurements[measurementIdx]),
null);
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance()
- .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath()));
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance()
+ .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath()));
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
@@ -318,12 +317,11 @@ public class DeletionFileNodeTest {
processorName + TsFileConstant.PATH_SEPARATOR + measurements[5]),
null);
- Pair<List<VirtualStorageGroupProcessor>, Map<VirtualStorageGroupProcessor, List<PartialPath>>>
- lockListAndProcessorToSeriesMapPair =
- StorageEngine.getInstance()
- .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath()));
- List<VirtualStorageGroupProcessor> lockList = lockListAndProcessorToSeriesMapPair.left;
- Map<VirtualStorageGroupProcessor, List<PartialPath>> processorToSeriesMap =
+ Pair<List<DataRegion>, Map<DataRegion, List<PartialPath>>> lockListAndProcessorToSeriesMapPair =
+ StorageEngine.getInstance()
+ .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath()));
+ List<DataRegion> lockList = lockListAndProcessorToSeriesMapPair.left;
+ Map<DataRegion, List<PartialPath>> processorToSeriesMap =
lockListAndProcessorToSeriesMapPair.right;
try {
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
index 1ce0722cde..29f318d4e9 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
@@ -78,7 +78,7 @@ public class StorageGroupProcessorTest {
private String systemDir = TestConstant.OUTPUT_DATA_DIR.concat("info");
private String deviceId = "root.vehicle.d0";
private String measurementId = "s0";
- private VirtualStorageGroupProcessor processor;
+ private DataRegion processor;
private QueryContext context = EnvironmentUtils.TEST_QUERY_CONTEXT;
@Before
@@ -923,7 +923,7 @@ public class StorageGroupProcessorTest {
config.setCloseTsFileIntervalAfterFlushing(prevCloseTsFileInterval);
}
- class DummySGP extends VirtualStorageGroupProcessor {
+ class DummySGP extends DataRegion {
DummySGP(String systemInfoDir, String storageGroupName) throws StorageGroupProcessorException {
super(systemInfoDir, "0", new TsFileFlushPolicy.DirectFlushPolicy(), storageGroupName);
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
index 29f6b787e9..c8a8dca6e2 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
@@ -81,7 +81,7 @@ public class TTLTest {
private String sg1 = "root.TTL_SG1";
private String sg2 = "root.TTL_SG2";
private long ttl = 12345;
- private VirtualStorageGroupProcessor virtualStorageGroupProcessor;
+ private DataRegion dataRegion;
private String s1 = "s1";
private String g1s1 = sg1 + IoTDBConstant.PATH_SEPARATOR + s1;
private long prevPartitionInterval;
@@ -96,7 +96,7 @@ public class TTLTest {
@After
public void tearDown() throws IOException, StorageEngineException {
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
EnvironmentUtils.cleanEnv();
IoTDBDescriptor.getInstance().getConfig().setPartitionInterval(prevPartitionInterval);
}
@@ -104,8 +104,8 @@ public class TTLTest {
private void createSchemas() throws MetadataException, StorageGroupProcessorException {
IoTDB.schemaProcessor.setStorageGroup(new PartialPath(sg1));
IoTDB.schemaProcessor.setStorageGroup(new PartialPath(sg2));
- virtualStorageGroupProcessor =
- new VirtualStorageGroupProcessor(
+ dataRegion =
+ new DataRegion(
IoTDBDescriptor.getInstance().getConfig().getSystemDir(),
sg1,
new DirectFlushPolicy(),
@@ -159,20 +159,20 @@ public class TTLTest {
plan.transferType();
// ok without ttl
- virtualStorageGroupProcessor.insert(plan);
+ dataRegion.insert(plan);
- virtualStorageGroupProcessor.setDataTTL(1000);
+ dataRegion.setDataTTL(1000);
// with ttl
plan.setTime(System.currentTimeMillis() - 1001);
boolean caught = false;
try {
- virtualStorageGroupProcessor.insert(plan);
+ dataRegion.insert(plan);
} catch (OutOfTTLException e) {
caught = true;
}
assertTrue(caught);
plan.setTime(System.currentTimeMillis() - 900);
- virtualStorageGroupProcessor.insert(plan);
+ dataRegion.insert(plan);
}
private void prepareData()
@@ -195,17 +195,17 @@ public class TTLTest {
// sequence data
for (int i = 1000; i < 2000; i++) {
plan.setTime(initTime - 2000 + i);
- virtualStorageGroupProcessor.insert(plan);
+ dataRegion.insert(plan);
if ((i + 1) % 300 == 0) {
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
}
// unsequence data
for (int i = 0; i < 1000; i++) {
plan.setTime(initTime - 2000 + i);
- virtualStorageGroupProcessor.insert(plan);
+ dataRegion.insert(plan);
if ((i + 1) % 300 == 0) {
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
}
}
}
@@ -218,7 +218,7 @@ public class TTLTest {
// files before ttl
QueryDataSource dataSource =
- virtualStorageGroupProcessor.query(
+ dataRegion.query(
Collections.singletonList(
SchemaTestUtils.getMeasurementPath(sg1 + TsFileConstant.PATH_SEPARATOR + s1)),
sg1,
@@ -230,11 +230,11 @@ public class TTLTest {
assertEquals(4, seqResource.size());
assertEquals(4, unseqResource.size());
- virtualStorageGroupProcessor.setDataTTL(500);
+ dataRegion.setDataTTL(500);
// files after ttl
dataSource =
- virtualStorageGroupProcessor.query(
+ dataRegion.query(
Collections.singletonList(
SchemaTestUtils.getMeasurementPath(sg1 + TsFileConstant.PATH_SEPARATOR + s1)),
sg1,
@@ -272,9 +272,9 @@ public class TTLTest {
// we cannot offer the exact number since when exactly ttl will be checked is unknown
assertTrue(cnt <= 1000);
- virtualStorageGroupProcessor.setDataTTL(0);
+ dataRegion.setDataTTL(0);
dataSource =
- virtualStorageGroupProcessor.query(
+ dataRegion.query(
Collections.singletonList(
SchemaTestUtils.getMeasurementPath(sg1 + TsFileConstant.PATH_SEPARATOR + s1)),
sg1,
@@ -293,7 +293,7 @@ public class TTLTest {
IllegalPathException {
prepareData();
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
// files before ttl
File seqDir = new File(DirectoryManager.getInstance().getNextFolderForSequenceFile(), sg1);
@@ -337,8 +337,8 @@ public class TTLTest {
} catch (InterruptedException e) {
e.printStackTrace();
}
- virtualStorageGroupProcessor.setDataTTL(500);
- virtualStorageGroupProcessor.checkFilesTTL();
+ dataRegion.setDataTTL(500);
+ dataRegion.checkFilesTTL();
// files after ttl
seqFiles = new ArrayList<>();
@@ -433,15 +433,15 @@ public class TTLTest {
throws WriteProcessException, QueryProcessException, IllegalPathException,
TriggerExecutionException {
prepareData();
- virtualStorageGroupProcessor.syncCloseAllWorkingTsFileProcessors();
+ dataRegion.syncCloseAllWorkingTsFileProcessors();
- assertEquals(4, virtualStorageGroupProcessor.getSequenceFileList().size());
- assertEquals(4, virtualStorageGroupProcessor.getUnSequenceFileList().size());
+ assertEquals(4, dataRegion.getSequenceFileList().size());
+ assertEquals(4, dataRegion.getUnSequenceFileList().size());
- virtualStorageGroupProcessor.setDataTTL(0);
- virtualStorageGroupProcessor.checkFilesTTL();
+ dataRegion.setDataTTL(0);
+ dataRegion.checkFilesTTL();
- assertEquals(0, virtualStorageGroupProcessor.getSequenceFileList().size());
- assertEquals(0, virtualStorageGroupProcessor.getUnSequenceFileList().size());
+ assertEquals(0, dataRegion.getSequenceFileList().size());
+ assertEquals(0, dataRegion.getUnSequenceFileList().size());
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/HashVirtualPartitionerTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/dataregion/HashVirtualPartitionerTest.java
similarity index 97%
rename from server/src/test/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/HashVirtualPartitionerTest.java
rename to server/src/test/java/org/apache/iotdb/db/engine/storagegroup/dataregion/HashVirtualPartitionerTest.java
index 6338c88025..ab0da5d7e1 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/virtualSg/HashVirtualPartitionerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/dataregion/HashVirtualPartitionerTest.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.db.engine.storagegroup.virtualSg;
+package org.apache.iotdb.db.engine.storagegroup.dataregion;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
diff --git a/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableFlushTimeTest.java b/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableFlushTimeTest.java
index 55cfb213aa..764b6aeee2 100644
--- a/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableFlushTimeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableFlushTimeTest.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.metadata.idtable;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -95,7 +95,7 @@ public class IDTableFlushTimeTest {
insertData(20);
- VirtualStorageGroupProcessor storageGroupProcessor =
+ DataRegion storageGroupProcessor =
StorageEngine.getInstance().getProcessor(new PartialPath("root.isp.d1"));
assertEquals(2, storageGroupProcessor.getSequenceFileList().size());
assertEquals(0, storageGroupProcessor.getUnSequenceFileList().size());
@@ -110,7 +110,7 @@ public class IDTableFlushTimeTest {
insertData(20);
- VirtualStorageGroupProcessor storageGroupProcessor =
+ DataRegion storageGroupProcessor =
StorageEngine.getInstance().getProcessor(new PartialPath("root.isp.d1"));
assertEquals(1, storageGroupProcessor.getSequenceFileList().size());
assertEquals(1, storageGroupProcessor.getUnSequenceFileList().size());
@@ -139,7 +139,7 @@ public class IDTableFlushTimeTest {
// unsequence
insertData(90);
- VirtualStorageGroupProcessor storageGroupProcessor =
+ DataRegion storageGroupProcessor =
StorageEngine.getInstance().getProcessor(new PartialPath("root.isp.d1"));
assertEquals(4, storageGroupProcessor.getSequenceFileList().size());
assertEquals(2, storageGroupProcessor.getUnSequenceFileList().size());
@@ -156,7 +156,7 @@ public class IDTableFlushTimeTest {
insertData(20);
insertData(120);
- VirtualStorageGroupProcessor storageGroupProcessor =
+ DataRegion storageGroupProcessor =
StorageEngine.getInstance().getProcessor(new PartialPath("root.isp.d1"));
assertEquals(
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
index 9a5c16c7a6..237bfb4dd2 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.db.mpp.execution;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
@@ -138,7 +138,7 @@ public class DataDriverTest {
new LimitOperator(
fragmentInstanceContext.getOperatorContexts().get(3), 250, timeJoinOperator);
- VirtualStorageGroupProcessor dataRegion = Mockito.mock(VirtualStorageGroupProcessor.class);
+ DataRegion dataRegion = Mockito.mock(DataRegion.class);
List<PartialPath> pathList = ImmutableList.of(measurementPath1, measurementPath2);
String deviceId = DATA_DRIVER_TEST_SG + ".device0";
diff --git a/server/src/test/java/org/apache/iotdb/db/sync/receiver/load/FileLoaderTest.java b/server/src/test/java/org/apache/iotdb/db/sync/receiver/load/FileLoaderTest.java
index e8786f4e25..0da22c8461 100644
--- a/server/src/test/java/org/apache/iotdb/db/sync/receiver/load/FileLoaderTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/sync/receiver/load/FileLoaderTest.java
@@ -22,8 +22,8 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -65,8 +65,8 @@ public class FileLoaderTest {
@Before
public void setUp() throws Exception {
- prevVirtualPartitionNum = IoTDBDescriptor.getInstance().getConfig().getVirtualStorageGroupNum();
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(1);
+ prevVirtualPartitionNum = IoTDBDescriptor.getInstance().getConfig().getDataRegionNum();
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(1);
IoTDBDescriptor.getInstance().getConfig().setSyncEnable(true);
EnvironmentUtils.envSetUp();
dataDir =
@@ -87,7 +87,7 @@ public class FileLoaderTest {
public void tearDown() throws IOException, StorageEngineException {
EnvironmentUtils.cleanEnv();
IoTDBDescriptor.getInstance().getConfig().setSyncEnable(false);
- IoTDBDescriptor.getInstance().getConfig().setVirtualStorageGroupNum(prevVirtualPartitionNum);
+ IoTDBDescriptor.getInstance().getConfig().setDataRegionNum(prevVirtualPartitionNum);
}
@Test
@@ -164,8 +164,7 @@ public class FileLoaderTest {
}
for (int i = 0; i < 3; i++) {
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
+ DataRegion processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
assertTrue(processor.getSequenceFileList().isEmpty());
assertTrue(processor.getUnSequenceFileList().isEmpty());
}
@@ -197,8 +196,7 @@ public class FileLoaderTest {
assertFalse(new File(getReceiverFolderFile(), SyncConstant.RECEIVER_DATA_FOLDER_NAME).exists());
Map<String, Set<String>> sequenceLoadedFileMap = new HashMap<>();
for (int i = 0; i < 3; i++) {
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
+ DataRegion processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
sequenceLoadedFileMap.putIfAbsent(SG_NAME + i, new HashSet<>());
assertEquals(10, processor.getSequenceFileList().size());
for (TsFileResource tsFileResource : processor.getSequenceFileList()) {
@@ -283,8 +281,7 @@ public class FileLoaderTest {
}
for (int i = 0; i < 3; i++) {
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
+ DataRegion processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
assertTrue(processor.getSequenceFileList().isEmpty());
assertTrue(processor.getUnSequenceFileList().isEmpty());
}
@@ -316,8 +313,7 @@ public class FileLoaderTest {
assertFalse(new File(getReceiverFolderFile(), SyncConstant.RECEIVER_DATA_FOLDER_NAME).exists());
Map<String, Set<String>> loadedFileMap = new HashMap<>();
for (int i = 0; i < 3; i++) {
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
+ DataRegion processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
loadedFileMap.putIfAbsent(SG_NAME + i, new HashSet<>());
assertEquals(25, processor.getSequenceFileList().size());
for (TsFileResource tsFileResource : processor.getSequenceFileList()) {
@@ -374,8 +370,7 @@ public class FileLoaderTest {
loadedFileMap.clear();
for (int i = 0; i < 3; i++) {
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
+ DataRegion processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
loadedFileMap.putIfAbsent(SG_NAME + i, new HashSet<>());
for (TsFileResource tsFileResource : processor.getSequenceFileList()) {
loadedFileMap.get(SG_NAME + i).add(tsFileResource.getTsFile().getAbsolutePath());
diff --git a/server/src/test/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzerTest.java b/server/src/test/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzerTest.java
index d6cf387cb8..4388baf5eb 100644
--- a/server/src/test/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/sync/receiver/recover/SyncReceiverLogAnalyzerTest.java
@@ -22,8 +22,8 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.storagegroup.DataRegion;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.engine.storagegroup.VirtualStorageGroupProcessor;
import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
@@ -154,8 +154,7 @@ public class SyncReceiverLogAnalyzerTest {
}
for (int i = 0; i < 3; i++) {
- VirtualStorageGroupProcessor processor =
- StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
+ DataRegion processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i));
assertTrue(processor.getSequenceFileList().isEmpty());
assertTrue(processor.getUnSequenceFileList().isEmpty());
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FilePathUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FilePathUtils.java
index a062e8b1ab..5e815c59f7 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FilePathUtils.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/FilePathUtils.java
@@ -72,7 +72,7 @@ public class FilePathUtils {
return pathSegments[pathSegments.length - 4];
}
- public static String getVirtualStorageGroupId(String tsFileAbsolutePath) {
+ public static String getDataRegionId(String tsFileAbsolutePath) {
String[] pathSegments = splitTsFilePath(tsFileAbsolutePath);
return pathSegments[pathSegments.length - 3];
}
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FilePathUtilsTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FilePathUtilsTest.java
index eb0fd69bdf..62a89c0031 100644
--- a/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FilePathUtilsTest.java
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/utils/FilePathUtilsTest.java
@@ -30,7 +30,7 @@ import java.io.IOException;
public class FilePathUtilsTest {
private static final String storageGroupName = "root.group_9";
- private static final String virtualSgName = "1";
+ private static final String dataRegionId = "1";
private static final long partitionId = 0;
private static final String tsFileName = "1611199237113-4-0.tsfile";
private static final String fullPath =
@@ -38,7 +38,7 @@ public class FilePathUtilsTest {
+ File.separator
+ storageGroupName
+ File.separator
- + virtualSgName
+ + dataRegionId
+ File.separator
+ partitionId
+ File.separator
@@ -75,8 +75,8 @@ public class FilePathUtilsTest {
@Test
public void getVirtualStorageGroupNameTest() {
- String tmpVirtualSgName = FilePathUtils.getVirtualStorageGroupId(tsFile.getAbsolutePath());
- Assert.assertEquals(virtualSgName, tmpVirtualSgName);
+ String tmpVirtualSgName = FilePathUtils.getDataRegionId(tsFile.getAbsolutePath());
+ Assert.assertEquals(dataRegionId, tmpVirtualSgName);
}
@Test
@@ -89,7 +89,7 @@ public class FilePathUtilsTest {
public void getTsFilePrefixPathTest() {
String tsFilePrefixPath = FilePathUtils.getTsFilePrefixPath(tsFile.getAbsolutePath());
String exceptPrefixPath =
- storageGroupName + File.separator + virtualSgName + File.separator + partitionId;
+ storageGroupName + File.separator + dataRegionId + File.separator + partitionId;
Assert.assertEquals(exceptPrefixPath, tsFilePrefixPath);
}