You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2023/09/15 10:02:29 UTC
[iotdb] branch rep_cp created (now f0cf96a2375)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch rep_cp
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at f0cf96a2375 Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113)
This branch includes the following new commits:
new f0cf96a2375 Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113)
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch rep_cp
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f0cf96a237566f408ffb5dd343e45d8133936775
Author: ZhangHongYin <46...@users.noreply.github.com>
AuthorDate: Fri Sep 15 16:45:19 2023 +0800
Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113)
---
.../manager/schema/ClusterSchemaManager.java | 63 +++++++++++++++++++++-
.../schemaengine/metric/SchemaEngineMemMetric.java | 24 ---------
.../schemaengine/metric/SchemaMetricManager.java | 3 +-
.../metric/SchemaRegionCachedMetric.java | 4 +-
.../schemaengine/metric/SchemaRegionMemMetric.java | 36 +++++++++----
.../schemaengine/schemaregion/ISchemaRegion.java | 2 +-
.../schemaregion/impl/SchemaRegionMemoryImpl.java | 4 +-
.../schemaregion/impl/SchemaRegionPBTreeImpl.java | 4 +-
.../dataregion/memtable/AbstractMemTable.java | 60 ++++++++++++++++++---
.../dataregion/memtable/IMemTable.java | 6 +++
.../dataregion/memtable/PrimitiveMemTable.java | 18 +++++--
.../dataregion/memtable/TsFileProcessor.java | 5 +-
.../dataregion/tsfile/TsFileResource.java | 8 +++
.../wal/recover/file/TsFilePlanRedoer.java | 4 +-
.../file/UnsealedTsFileRecoverPerformer.java | 10 ++--
.../rescon/memory/MemTableManager.java | 8 +--
.../dataregion/memtable/MemTableFlushTaskTest.java | 4 +-
.../dataregion/memtable/MemtableBenchmark.java | 4 +-
.../dataregion/memtable/PrimitiveMemTableTest.java | 16 +++---
.../wal/checkpoint/CheckpointManagerTest.java | 8 ++-
.../dataregion/wal/io/CheckpointFileTest.java | 8 ++-
.../dataregion/wal/node/WALEntryHandlerTest.java | 20 +++----
.../dataregion/wal/node/WALNodeTest.java | 10 ++--
.../wal/recover/WALRecoverManagerTest.java | 9 ++--
.../wal/utils/WALInsertNodeCacheTest.java | 10 ++--
.../iotdb/commons/service/metric/enums/Metric.java | 1 +
26 files changed, 251 insertions(+), 98 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
index dfacde2296c..1f7c3417415 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -27,6 +27,8 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.service.metric.MetricService;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
@@ -82,6 +84,8 @@ import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUpdateType;
import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUtil;
import org.apache.iotdb.db.schemaengine.template.alter.TemplateExtendInfo;
import org.apache.iotdb.db.utils.SchemaUtils;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -120,6 +124,9 @@ public class ClusterSchemaManager {
private static final String CONSENSUS_WRITE_ERROR =
"Failed in the write API executing the consensus layer due to: ";
+ private static final MetricService metricService = MetricService.getInstance();
+ private static final String DATA = "data";
+ private static final String SCHEMA = "schema";
public ClusterSchemaManager(
IManager configManager,
@@ -166,6 +173,25 @@ public class ClusterSchemaManager {
MetricService.getInstance(), configManager, databaseSchemaPlan.getSchema().getName());
// Adjust the maximum RegionGroup number of each Database
adjustMaxRegionGroupNum();
+ // Add database replication factor metrics
+ metricService
+ .getOrCreateGauge(
+ Metric.REPLICATION_FACTOR.toString(),
+ MetricLevel.CORE,
+ Tag.TYPE.toString(),
+ DATA,
+ Tag.DATABASE.toString(),
+ databaseSchemaPlan.getSchema().getName())
+ .set(databaseSchemaPlan.getSchema().dataReplicationFactor);
+ metricService
+ .getOrCreateGauge(
+ Metric.REPLICATION_FACTOR.toString(),
+ MetricLevel.CORE,
+ Tag.TYPE.toString(),
+ SCHEMA,
+ Tag.DATABASE.toString(),
+ databaseSchemaPlan.getSchema().getName())
+ .set(databaseSchemaPlan.getSchema().schemaReplicationFactor);
} catch (ConsensusException e) {
LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
result = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
@@ -233,7 +259,27 @@ public class ClusterSchemaManager {
// Alter DatabaseSchema
try {
- return getConsensusManager().write(databaseSchemaPlan);
+ result = getConsensusManager().write(databaseSchemaPlan);
+ // Alter database replication factor metrics
+ metricService
+ .getOrCreateGauge(
+ Metric.REPLICATION_FACTOR.toString(),
+ MetricLevel.CORE,
+ Tag.TYPE.toString(),
+ DATA,
+ Tag.DATABASE.toString(),
+ databaseSchemaPlan.getSchema().getName())
+ .set(databaseSchemaPlan.getSchema().dataReplicationFactor);
+ metricService
+ .getOrCreateGauge(
+ Metric.REPLICATION_FACTOR.toString(),
+ MetricLevel.CORE,
+ Tag.TYPE.toString(),
+ SCHEMA,
+ Tag.DATABASE.toString(),
+ databaseSchemaPlan.getSchema().getName())
+ .set(databaseSchemaPlan.getSchema().schemaReplicationFactor);
+ return result;
} catch (ConsensusException e) {
LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
result = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
@@ -247,6 +293,21 @@ public class ClusterSchemaManager {
TSStatus result;
try {
result = getConsensusManager().write(deleteDatabasePlan);
+ // Remove database replication factor metric
+ metricService.remove(
+ MetricType.GAUGE,
+ Metric.REPLICATION_FACTOR.toString(),
+ Tag.TYPE.toString(),
+ DATA,
+ Tag.DATABASE.toString(),
+ deleteDatabasePlan.getName());
+ metricService.remove(
+ MetricType.GAUGE,
+ Metric.REPLICATION_FACTOR.toString(),
+ Tag.TYPE.toString(),
+ SCHEMA,
+ Tag.DATABASE.toString(),
+ deleteDatabasePlan.getName());
} catch (ConsensusException e) {
LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
result = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java
index 3f5c6ec7211..88080782e16 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaEngineMemMetric.java
@@ -33,7 +33,6 @@ import org.apache.iotdb.metrics.utils.MetricType;
public class SchemaEngineMemMetric implements ISchemaEngineMetric {
- private static final String TIME_SERES_CNT = "timeSeries";
public static final String SCHEMA_REGION_TOTAL_USAGE = "schema_region_total_usage";
private static final String TOTAL_MEM_USAGE = "schema_region_total_mem_usage";
private static final String MEM_CAPACITY = "schema_region_mem_capacity";
@@ -42,8 +41,6 @@ public class SchemaEngineMemMetric implements ISchemaEngineMetric {
private static final String SCHEMA_CONSENSUS = "schema_region_consensus";
private static final String SCHEMA_ENGINE_MODE = "schema_engine_mode";
- private static final String TEMPLATE_SERIES_CNT = "template_series_cnt";
-
private final MemSchemaEngineStatistics engineStatistics;
public SchemaEngineMemMetric(MemSchemaEngineStatistics engineStatistics) {
@@ -52,20 +49,6 @@ public class SchemaEngineMemMetric implements ISchemaEngineMetric {
@Override
public void bindTo(AbstractMetricService metricService) {
- metricService.createAutoGauge(
- Metric.QUANTITY.toString(),
- MetricLevel.CORE,
- engineStatistics,
- ISchemaEngineStatistics::getTotalSeriesNumber,
- Tag.NAME.toString(),
- TIME_SERES_CNT);
- metricService.createAutoGauge(
- Metric.QUANTITY.toString(),
- MetricLevel.IMPORTANT,
- engineStatistics,
- ISchemaEngineStatistics::getTemplateSeriesNumber,
- Tag.NAME.toString(),
- TEMPLATE_SERIES_CNT);
metricService.createAutoGauge(
Metric.MEM.toString(),
MetricLevel.IMPORTANT,
@@ -111,13 +94,6 @@ public class SchemaEngineMemMetric implements ISchemaEngineMetric {
@Override
public void unbindFrom(AbstractMetricService metricService) {
- metricService.remove(
- MetricType.AUTO_GAUGE, Metric.QUANTITY.toString(), Tag.NAME.toString(), TIME_SERES_CNT);
- metricService.remove(
- MetricType.AUTO_GAUGE,
- Metric.QUANTITY.toString(),
- Tag.NAME.toString(),
- TEMPLATE_SERIES_CNT);
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.MEM.toString(),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaMetricManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaMetricManager.java
index 30e38b37292..32ede440dbd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaMetricManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaMetricManager.java
@@ -46,7 +46,8 @@ public class SchemaMetricManager {
}
public void createSchemaRegionMetric(ISchemaRegion schemaRegion) {
- ISchemaRegionMetric schemaRegionMetric = schemaRegion.createSchemaRegionMetric();
+ ISchemaRegionMetric schemaRegionMetric =
+ schemaRegion.createSchemaRegionMetric(schemaRegion.getDatabaseFullPath());
schemaRegionMetricMap.put(schemaRegion.getSchemaRegionId().getId(), schemaRegionMetric);
MetricService.getInstance().addMetricSet(schemaRegionMetric);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionCachedMetric.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionCachedMetric.java
index 9d1f892f893..079b1d7dab4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionCachedMetric.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionCachedMetric.java
@@ -43,10 +43,10 @@ public class SchemaRegionCachedMetric implements ISchemaRegionMetric {
// MemSchemaRegionMetric is a subset of CachedSchemaRegionMetric
private final SchemaRegionMemMetric memSchemaRegionMetric;
- public SchemaRegionCachedMetric(CachedSchemaRegionStatistics regionStatistics) {
+ public SchemaRegionCachedMetric(CachedSchemaRegionStatistics regionStatistics, String database) {
this.regionStatistics = regionStatistics;
this.regionTagValue = String.format("SchemaRegion[%d]", regionStatistics.getSchemaRegionId());
- this.memSchemaRegionMetric = new SchemaRegionMemMetric(regionStatistics);
+ this.memSchemaRegionMetric = new SchemaRegionMemMetric(regionStatistics, database);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java
index 8e0bee1c156..0e0f324087d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/metric/SchemaRegionMemMetric.java
@@ -35,10 +35,12 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
private final MemSchemaRegionStatistics regionStatistics;
private final String regionTagValue;
+ private final String database;
- public SchemaRegionMemMetric(MemSchemaRegionStatistics regionStatistics) {
+ public SchemaRegionMemMetric(MemSchemaRegionStatistics regionStatistics, String database) {
this.regionStatistics = regionStatistics;
this.regionTagValue = String.format("SchemaRegion[%d]", regionStatistics.getSchemaRegionId());
+ this.database = database;
}
@Override
@@ -51,7 +53,9 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
Tag.NAME.toString(),
MEM_USAGE,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
metricService.createAutoGauge(
Metric.SCHEMA_REGION.toString(),
MetricLevel.IMPORTANT,
@@ -60,7 +64,9 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
Tag.NAME.toString(),
SERIES_CNT,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
metricService.createAutoGauge(
Metric.SCHEMA_REGION.toString(),
MetricLevel.IMPORTANT,
@@ -69,7 +75,9 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
Tag.NAME.toString(),
TEMPLATE_CNT,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
metricService.createAutoGauge(
Metric.SCHEMA_REGION.toString(),
MetricLevel.IMPORTANT,
@@ -78,7 +86,9 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
Tag.NAME.toString(),
TEMPLATE_SERIES_CNT,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
}
@Override
@@ -89,27 +99,35 @@ public class SchemaRegionMemMetric implements ISchemaRegionMetric {
Tag.NAME.toString(),
MEM_USAGE,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SCHEMA_REGION.toString(),
Tag.NAME.toString(),
SERIES_CNT,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SCHEMA_REGION.toString(),
Tag.NAME.toString(),
TEMPLATE_CNT,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SCHEMA_REGION.toString(),
Tag.NAME.toString(),
TEMPLATE_SERIES_CNT,
Tag.REGION.toString(),
- regionTagValue);
+ regionTagValue,
+ Tag.DATABASE.toString(),
+ database);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java
index 815685c851e..b724d82de32 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java
@@ -83,7 +83,7 @@ public interface ISchemaRegion {
@TestOnly
ISchemaRegionStatistics getSchemaRegionStatistics();
- ISchemaRegionMetric createSchemaRegionMetric();
+ ISchemaRegionMetric createSchemaRegionMetric(String database);
// endregion
// region Interfaces for schema region Info query and operation
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
index 3a49f596f9e..aec539bdea8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
@@ -288,8 +288,8 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion {
}
@Override
- public ISchemaRegionMetric createSchemaRegionMetric() {
- return new SchemaRegionMemMetric(regionStatistics);
+ public ISchemaRegionMetric createSchemaRegionMetric(String database) {
+ return new SchemaRegionMemMetric(regionStatistics, database);
}
/**
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
index d7f008ab59d..6a1b6221145 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
@@ -329,8 +329,8 @@ public class SchemaRegionPBTreeImpl implements ISchemaRegion {
}
@Override
- public ISchemaRegionMetric createSchemaRegionMetric() {
- return new SchemaRegionCachedMetric(regionStatistics);
+ public ISchemaRegionMetric createSchemaRegionMetric(String database) {
+ return new SchemaRegionCachedMetric(regionStatistics, database);
}
/** Init from metadata log file. */
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
index bdba167c5bb..f3264c09738 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
@@ -65,7 +65,8 @@ public abstract class AbstractMemTable implements IMemTable {
private final Map<IDeviceID, IWritableMemChunkGroup> memTableMap;
/**
- * The initial value is true because we want calculate the text data size when recover memTable.
+ * The initial value is true because we want to calculate the text data size when recover
+ * memTable.
*/
protected boolean disableMemControl = true;
@@ -95,13 +96,27 @@ public abstract class AbstractMemTable implements IMemTable {
private final long createdTime = System.currentTimeMillis();
+ private String database;
+ private String dataRegionId;
+
private static final String METRIC_POINT_IN = "pointsIn";
protected AbstractMemTable() {
+ this.database = null;
+ this.dataRegionId = null;
+ this.memTableMap = new HashMap<>();
+ }
+
+ protected AbstractMemTable(String database, String dataRegionId) {
+ this.database = database;
+ this.dataRegionId = dataRegionId;
this.memTableMap = new HashMap<>();
}
- protected AbstractMemTable(Map<IDeviceID, IWritableMemChunkGroup> memTableMap) {
+ protected AbstractMemTable(
+ String database, String dataRegionId, Map<IDeviceID, IWritableMemChunkGroup> memTableMap) {
+ this.database = database;
+ this.dataRegionId = dataRegionId;
this.memTableMap = memTableMap;
}
@@ -194,7 +209,11 @@ public abstract class AbstractMemTable implements IMemTable {
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
- METRIC_POINT_IN);
+ METRIC_POINT_IN,
+ Tag.DATABASE.toString(),
+ database,
+ Tag.REGION.toString(),
+ dataRegionId);
}
@Override
@@ -233,7 +252,11 @@ public abstract class AbstractMemTable implements IMemTable {
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
- METRIC_POINT_IN);
+ METRIC_POINT_IN,
+ Tag.DATABASE.toString(),
+ database,
+ Tag.REGION.toString(),
+ dataRegionId);
}
@Override
@@ -252,7 +275,11 @@ public abstract class AbstractMemTable implements IMemTable {
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
- METRIC_POINT_IN);
+ METRIC_POINT_IN,
+ Tag.DATABASE.toString(),
+ database,
+ Tag.REGION.toString(),
+ dataRegionId);
} catch (RuntimeException e) {
throw new WriteProcessException(e);
}
@@ -274,7 +301,11 @@ public abstract class AbstractMemTable implements IMemTable {
Metric.QUANTITY.toString(),
MetricLevel.CORE,
Tag.NAME.toString(),
- METRIC_POINT_IN);
+ METRIC_POINT_IN,
+ Tag.DATABASE.toString(),
+ database,
+ Tag.REGION.toString(),
+ dataRegionId);
} catch (RuntimeException e) {
throw new WriteProcessException(e);
}
@@ -645,6 +676,7 @@ public abstract class AbstractMemTable implements IMemTable {
if (isSignal) {
memTable = new NotifyFlushMemTable();
} else {
+ // database will be updated when deserialize
PrimitiveMemTable primitiveMemTable = new PrimitiveMemTable();
primitiveMemTable.deserialize(stream);
memTable = primitiveMemTable;
@@ -652,4 +684,20 @@ public abstract class AbstractMemTable implements IMemTable {
return memTable;
}
}
+
+ @Override
+ public String getDatabase() {
+ return database;
+ }
+
+ @Override
+ public String getDataRegionId() {
+ return dataRegionId;
+ }
+
+ @Override
+ public void setDatabaseAndDataRegionId(String database, String dataRegionId) {
+ this.database = database;
+ this.dataRegionId = dataRegionId;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
index b4b0204080d..420422c8537 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
@@ -169,4 +169,10 @@ public interface IMemTable extends WALEntryValue {
void setFlushStatus(FlushStatus flushStatus);
Map<String, Long> getMaxTime();
+
+ String getDatabase();
+
+ String getDataRegionId();
+
+ void setDatabaseAndDataRegionId(String database, String dataRegionId);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTable.java
index d4bf9c18743..f8e625a3841 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTable.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTable.java
@@ -23,22 +23,30 @@ import java.util.HashMap;
import java.util.Map;
public class PrimitiveMemTable extends AbstractMemTable {
+ // this constructor only used when deserialize
+ public PrimitiveMemTable() {
+ super();
+ }
- public PrimitiveMemTable() {}
+ public PrimitiveMemTable(String database, String dataRegionId) {
+ super(database, dataRegionId);
+ }
- public PrimitiveMemTable(boolean enableMemControl) {
+ public PrimitiveMemTable(String database, String dataRegionId, boolean enableMemControl) {
+ super(database, dataRegionId);
this.disableMemControl = !enableMemControl;
}
- public PrimitiveMemTable(Map<IDeviceID, IWritableMemChunkGroup> memTableMap) {
- super(memTableMap);
+ public PrimitiveMemTable(
+ String database, String dataRegionId, Map<IDeviceID, IWritableMemChunkGroup> memTableMap) {
+ super(database, dataRegionId, memTableMap);
}
@Override
public IMemTable copy() {
Map<IDeviceID, IWritableMemChunkGroup> newMap = new HashMap<>(getMemTableMap());
- return new PrimitiveMemTable(newMap);
+ return new PrimitiveMemTable(getDatabase(), getDataRegionId(), newMap);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 3cd09e73e92..eb62cd52fd2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -306,7 +306,10 @@ public class TsFileProcessor {
}
private void createNewWorkingMemTable() throws WriteProcessException {
- workMemTable = MemTableManager.getInstance().getAvailableMemTable(storageGroupName);
+ workMemTable =
+ MemTableManager.getInstance()
+ .getAvailableMemTable(
+ storageGroupName, dataRegionInfo.getDataRegion().getDataRegionId());
walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index bcc4ff7664e..bbd3a17d017 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -1191,4 +1191,12 @@ public class TsFileResource {
public ProgressIndex getMaxProgressIndex() {
return maxProgressIndex == null ? MinimumProgressIndex.INSTANCE : maxProgressIndex;
}
+
+ public String getDatabaseName() {
+ return file.getParentFile().getParentFile().getParentFile().getName();
+ }
+
+ public String getDataRegionId() {
+ return file.getParentFile().getParentFile().getName();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
index d7a292cf36c..f862282d411 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
@@ -45,11 +45,13 @@ public class TsFilePlanRedoer {
private final boolean sequence;
// store data when redoing logs
- private IMemTable recoveryMemTable = new PrimitiveMemTable();
+ private IMemTable recoveryMemTable;
public TsFilePlanRedoer(TsFileResource tsFileResource, boolean sequence) {
this.tsFileResource = tsFileResource;
this.sequence = sequence;
+ this.recoveryMemTable =
+ new PrimitiveMemTable(tsFileResource.getDatabaseName(), tsFileResource.getDataRegionId());
}
void redoDelete(DeleteDataNode deleteDataNode) throws IOException {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java
index 742e996fd03..13a97b208af 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java
@@ -69,12 +69,16 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform
private final TsFilePlanRedoer walRedoer;
// trace result of this recovery
private final WALRecoverListener recoverListener;
+ private final String databaseName;
+ private final String dataRegionId;
public UnsealedTsFileRecoverPerformer(
TsFileResource tsFileResource,
boolean sequence,
Consumer<UnsealedTsFileRecoverPerformer> callbackAfterUnsealedTsFileRecovered) {
super(tsFileResource);
+ this.databaseName = tsFileResource.getDatabaseName();
+ this.dataRegionId = tsFileResource.getDataRegionId();
this.sequence = sequence;
this.callbackAfterUnsealedTsFileRecovered = callbackAfterUnsealedTsFileRecovered;
this.walRedoer = new TsFilePlanRedoer(tsFileResource, sequence);
@@ -195,6 +199,8 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform
if (!memTable.isSignalMemTable()) {
walRedoer.resetRecoveryMemTable(memTable);
}
+ // update memtable's database and dataRegionId
+ memTable.setDatabaseAndDataRegionId(databaseName, dataRegionId);
break;
case INSERT_ROW_NODE:
case INSERT_TABLET_NODE:
@@ -234,10 +240,6 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform
// flush memTable
try {
if (!recoveryMemTable.isEmpty() && recoveryMemTable.getSeriesNumber() != 0) {
- String dataRegionId =
- tsFileResource.getTsFile().getParentFile().getParentFile().getName();
- String databaseName =
- tsFileResource.getTsFile().getParentFile().getParentFile().getParentFile().getName();
MemTableFlushTask tableFlushTask =
new MemTableFlushTask(
recoveryMemTable,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/MemTableManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/MemTableManager.java
index 0359aec79a1..5f44e11099e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/MemTableManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/MemTableManager.java
@@ -43,16 +43,16 @@ public class MemTableManager {
return InstanceHolder.INSTANCE;
}
- public synchronized IMemTable getAvailableMemTable(String storageGroup)
+ public synchronized IMemTable getAvailableMemTable(String storageGroup, String dataRegionId)
throws WriteProcessException {
if (CONFIG.isEnableMemControl()) {
currentMemtableNumber++;
- return new PrimitiveMemTable(CONFIG.isEnableMemControl());
+ return new PrimitiveMemTable(storageGroup, dataRegionId, CONFIG.isEnableMemControl());
}
if (!reachMaxMemtableNumber()) {
currentMemtableNumber++;
- return new PrimitiveMemTable();
+ return new PrimitiveMemTable(storageGroup, dataRegionId);
}
// wait until the total number of memtable is less than the system capacity
@@ -60,7 +60,7 @@ public class MemTableManager {
while (true) {
if (!reachMaxMemtableNumber()) {
currentMemtableNumber++;
- return new PrimitiveMemTable();
+ return new PrimitiveMemTable(storageGroup, dataRegionId);
}
try {
wait(WAIT_TIME);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
index 907ff94319d..059ce487c17 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
@@ -41,7 +41,7 @@ public class MemTableFlushTaskTest {
private RestorableTsFileIOWriter writer;
private String storageGroup = "storage_group1";
- private String dataRegionId = "DataRegion-1";
+ private String dataRegionId = "1";
private String filePath =
TestConstant.OUTPUT_DATA_DIR.concat("testUnsealedTsFileProcessor.tsfile");
private IMemTable memTable;
@@ -52,7 +52,7 @@ public class MemTableFlushTaskTest {
public void setUp() throws Exception {
EnvironmentUtils.envSetUp();
writer = new RestorableTsFileIOWriter(FSFactoryProducer.getFSFactory().getFile(filePath));
- memTable = new PrimitiveMemTable();
+ memTable = new PrimitiveMemTable(storageGroup, dataRegionId);
}
@After
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemtableBenchmark.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemtableBenchmark.java
index ca697b3257c..1338b067333 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemtableBenchmark.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemtableBenchmark.java
@@ -29,6 +29,8 @@ import java.util.Collections;
/** Memtable insert benchmark. Bench the Memtable and get its performance. */
public class MemtableBenchmark {
+ private static String database = "root.test";
+ private static String dataRegionId = "1";
private static String deviceId = "d0";
private static int numOfMeasurement = 10000;
private static int numOfPoint = 1000;
@@ -43,7 +45,7 @@ public class MemtableBenchmark {
}
public static void main(String[] args) throws IllegalPathException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
final long startTime = System.currentTimeMillis();
// cpu not locality
for (int i = 0; i < numOfPoint; i++) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java
index 86d55cc771c..9bafd6f09d0 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/PrimitiveMemTableTest.java
@@ -60,6 +60,8 @@ import static org.junit.Assert.assertEquals;
public class PrimitiveMemTableTest {
+ String database = "root.test";
+ String dataRegionId = "1";
double delta;
@Before
@@ -114,7 +116,7 @@ public class PrimitiveMemTableTest {
@Test
public void simpleTest() throws IOException, QueryProcessException, MetadataException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
int count = 10;
String deviceId = "d1";
String[] measurementId = new String[count];
@@ -161,7 +163,7 @@ public class PrimitiveMemTableTest {
@Test
public void totalSeriesNumberTest() throws IOException, QueryProcessException, MetadataException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
int count = 10;
String deviceId = "d1";
String[] measurementId = new String[count];
@@ -211,7 +213,7 @@ public class PrimitiveMemTableTest {
@Test
public void queryWithDeletionTest() throws IOException, QueryProcessException, MetadataException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
int count = 10;
String deviceId = "d1";
String[] measurementId = new String[count];
@@ -265,7 +267,7 @@ public class PrimitiveMemTableTest {
@Test
public void queryAlignChuckWithDeletionTest()
throws IOException, QueryProcessException, MetadataException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
int count = 10;
String deviceId = "d1";
String[] measurementId = new String[count];
@@ -426,7 +428,7 @@ public class PrimitiveMemTableTest {
@Test
public void testFloatType() throws IOException, QueryProcessException, MetadataException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
String deviceId = "d1";
int size = 100;
write(memTable, deviceId, "s1", TSDataType.FLOAT, TSEncoding.RLE, size);
@@ -435,7 +437,7 @@ public class PrimitiveMemTableTest {
@Test
public void testAllType()
throws IOException, QueryProcessException, MetadataException, WriteProcessException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
int count = 10;
String deviceId = "d1";
String[] measurementId = new String[count];
@@ -539,7 +541,7 @@ public class PrimitiveMemTableTest {
@Test
public void testSerializeSize()
throws IOException, QueryProcessException, MetadataException, WriteProcessException {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(database, dataRegionId);
int count = 10;
String deviceId = "d1";
String[] measurementId = new String[count];
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManagerTest.java
index a5039fd4d07..ab9b83ce475 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManagerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManagerTest.java
@@ -48,6 +48,8 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class CheckpointManagerTest {
+ private static final String database = "root.test";
+ private static final String dataRegionId = "1";
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final String identifier = String.valueOf(Integer.MAX_VALUE);
private static final String logDirectory = TestConstant.BASE_OUTPUT_PATH.concat("wal-test");
@@ -97,7 +99,8 @@ public class CheckpointManagerTest {
() -> {
String tsFilePath = logDirectory + File.separator + versionId + ".tsfile";
MemTableInfo memTableInfo =
- new MemTableInfo(new PrimitiveMemTable(), tsFilePath, versionId);
+ new MemTableInfo(
+ new PrimitiveMemTable(database, dataRegionId), tsFilePath, versionId);
versionId2memTableId.put(versionId, memTableInfo.getMemTableId());
checkpointManager.makeCreateMemTableCP(memTableInfo);
if (versionId < memTablesNum / 2) {
@@ -133,7 +136,8 @@ public class CheckpointManagerTest {
while (size < config.getCheckpointFileSizeThresholdInByte()) {
++versionId;
String tsFilePath = logDirectory + File.separator + versionId + ".tsfile";
- MemTableInfo memTableInfo = new MemTableInfo(new PrimitiveMemTable(), tsFilePath, versionId);
+ MemTableInfo memTableInfo =
+ new MemTableInfo(new PrimitiveMemTable(database, dataRegionId), tsFilePath, versionId);
versionId2memTableId.put(versionId, memTableInfo.getMemTableId());
Checkpoint checkpoint =
new Checkpoint(
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointFileTest.java
index 78e898c08c2..ed7b7715e35 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointFileTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointFileTest.java
@@ -41,6 +41,8 @@ import static org.junit.Assert.assertEquals;
public class CheckpointFileTest {
private final File checkpointFile =
new File(TestConstant.BASE_OUTPUT_PATH.concat("_0.checkpoint"));
+ private static final String database = "root.test";
+ private static final String dataRegionId = "1";
@Before
public void setUp() throws Exception {
@@ -58,7 +60,8 @@ public class CheckpointFileTest {
@Test
public void testReadNormalFile() throws IOException {
- MemTableInfo fakeMemTableInfo = new MemTableInfo(new PrimitiveMemTable(), "fake.tsfile", 0);
+ MemTableInfo fakeMemTableInfo =
+ new MemTableInfo(new PrimitiveMemTable(database, dataRegionId), "fake.tsfile", 0);
List<Checkpoint> expectedCheckpoints = new ArrayList<>();
expectedCheckpoints.add(
new Checkpoint(CheckpointType.GLOBAL_MEMORY_TABLE_INFO, Collections.emptyList()));
@@ -101,7 +104,8 @@ public class CheckpointFileTest {
@Test
public void testReadBrokenFile() throws IOException {
- MemTableInfo fakeMemTableInfo = new MemTableInfo(new PrimitiveMemTable(), "fake.tsfile", 0);
+ MemTableInfo fakeMemTableInfo =
+ new MemTableInfo(new PrimitiveMemTable(database, dataRegionId), "fake.tsfile", 0);
List<Checkpoint> expectedCheckpoints = new ArrayList<>();
expectedCheckpoints.add(
new Checkpoint(CheckpointType.GLOBAL_MEMORY_TABLE_INFO, Collections.emptyList()));
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALEntryHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALEntryHandlerTest.java
index 39dda5f68c4..4dd69fb0162 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALEntryHandlerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALEntryHandlerTest.java
@@ -67,7 +67,9 @@ public class WALEntryHandlerTest {
private static final String logDirectory2 =
TestConstant.BASE_OUTPUT_PATH.concat("wal-test" + identifier2);
- private static final String devicePath = "root.test_sg.test_d";
+ private static final String databasePath = "root.test_sg";
+ private static final String devicePath = databasePath + ".test_d";
+ private static final String dataRegionId = "1";
private WALMode prevMode;
private boolean prevIsClusterMode;
private WALNode walNode1;
@@ -98,7 +100,7 @@ public class WALEntryHandlerTest {
@Test(expected = MemTablePinException.class)
public void pinDeletedMemTable() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
WALFlushListener flushListener =
walNode1.log(
@@ -111,7 +113,7 @@ public class WALEntryHandlerTest {
@Test
public void pinMemTable() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
InsertRowNode node1 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(1);
@@ -141,7 +143,7 @@ public class WALEntryHandlerTest {
@Test(expected = MemTablePinException.class)
public void unpinDeletedMemTable() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
WALFlushListener flushListener =
walNode1.log(
@@ -154,7 +156,7 @@ public class WALEntryHandlerTest {
@Test
public void unpinFlushedMemTable() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
WALFlushListener flushListener =
walNode1.log(
@@ -178,7 +180,7 @@ public class WALEntryHandlerTest {
@Test
public void unpinMemTable() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
InsertRowNode node1 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(1);
@@ -207,7 +209,7 @@ public class WALEntryHandlerTest {
@Test
public void getUnFlushedValue() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
InsertRowNode node1 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(1);
@@ -221,7 +223,7 @@ public class WALEntryHandlerTest {
@Test
public void getFlushedValue() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode1.onMemTableCreated(memTable, logDirectory1 + "/" + "fake.tsfile");
InsertRowNode node1 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(1);
@@ -247,7 +249,7 @@ public class WALEntryHandlerTest {
String logDirectory = i % 2 == 0 ? logDirectory1 : logDirectory2;
Callable<Void> writeTask =
() -> {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode.onMemTableCreated(memTable, logDirectory + "/" + "fake.tsfile");
List<WALFlushListener> walFlushListeners = new ArrayList<>();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java
index 4d6d30883fe..7e9211ae3ca 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeTest.java
@@ -68,7 +68,9 @@ public class WALNodeTest {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final String identifier = String.valueOf(Integer.MAX_VALUE);
private static final String logDirectory = TestConstant.BASE_OUTPUT_PATH.concat("wal-test");
- private static final String devicePath = "root.test_sg.test_d";
+ private static final String databasePath = "root.test_sg";
+ private static final String devicePath = databasePath + ".test_d";
+ private static final String dataRegionId = "1";
private WALMode prevMode;
private boolean prevIsClusterMode;
private WALNode walNode;
@@ -226,7 +228,7 @@ public class WALNodeTest {
for (int i = 0; i < memTablesNum; ++i) {
Callable<Void> writeTask =
() -> {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
long memTableId = memTable.getMemTableId();
String tsFilePath = logDirectory + File.separator + memTableId + ".tsfile";
long firstFileVersionId = walNode.getCurrentLogVersion();
@@ -260,7 +262,7 @@ public class WALNodeTest {
List<WALFlushListener> walFlushListeners = new ArrayList<>();
// write until log is rolled
long time = 0;
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
long memTableId = memTable.getMemTableId();
String tsFilePath = logDirectory + File.separator + memTableId + ".tsfile";
walNode.onMemTableCreated(memTable, tsFilePath);
@@ -273,7 +275,7 @@ public class WALNodeTest {
walFlushListeners.add(walFlushListener);
}
walNode.onMemTableFlushed(memTable);
- walNode.onMemTableCreated(new PrimitiveMemTable(), tsFilePath);
+ walNode.onMemTableCreated(new PrimitiveMemTable(databasePath, dataRegionId), tsFilePath);
// check existence of _0-0-0.wal file and _1-0-1.wal file
assertTrue(
new File(
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java
index 1a0d3a07d8c..8777464abf4 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java
@@ -88,6 +88,7 @@ public class WALRecoverManagerTest {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
private static final String SG_NAME = "root.recover_sg";
+ private static final String DATA_REGION_ID = "1";
private static final String DEVICE1_NAME = SG_NAME.concat(".d1");
private static final String DEVICE2_NAME = SG_NAME.concat(".d2");
private static final String FILE_WITH_WAL_NAME =
@@ -151,7 +152,7 @@ public class WALRecoverManagerTest {
List<Future<Void>> futures = new ArrayList<>();
long firstWALVersionId = walBuffer.getCurrentWALFileVersion();
for (int i = 0; i < threadsNum; ++i) {
- IMemTable fakeMemTable = new PrimitiveMemTable();
+ IMemTable fakeMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID);
long memTableId = fakeMemTable.getMemTableId();
Callable<Void> writeTask =
() -> {
@@ -185,7 +186,7 @@ public class WALRecoverManagerTest {
Thread.sleep(1_000);
// write normal .wal files
long firstValidVersionId = walBuffer.getCurrentWALFileVersion();
- IMemTable targetMemTable = new PrimitiveMemTable();
+ IMemTable targetMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID);
WALEntry walEntry =
new WALInfoEntry(targetMemTable.getMemTableId(), getInsertRowNode(DEVICE2_NAME, 4L), true);
walBuffer.write(walEntry);
@@ -210,7 +211,7 @@ public class WALRecoverManagerTest {
List<Future<Void>> futures = new ArrayList<>();
long firstWALVersionId = walBuffer.getCurrentWALFileVersion();
for (int i = 0; i < threadsNum; ++i) {
- IMemTable fakeMemTable = new PrimitiveMemTable();
+ IMemTable fakeMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID);
long memTableId = fakeMemTable.getMemTableId();
Callable<Void> writeTask =
() -> {
@@ -244,7 +245,7 @@ public class WALRecoverManagerTest {
Thread.sleep(1_000);
// write normal .wal files
long firstValidVersionId = walBuffer.getCurrentWALFileVersion();
- IMemTable targetMemTable = new PrimitiveMemTable();
+ IMemTable targetMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID);
InsertRowNode insertRowNode = getInsertRowNode(DEVICE2_NAME, 4L);
targetMemTable.insert(insertRowNode);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCacheTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCacheTest.java
index 1518ffc4dcf..18817591195 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCacheTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALInsertNodeCacheTest.java
@@ -46,7 +46,9 @@ public class WALInsertNodeCacheTest {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static final String identifier = String.valueOf(Integer.MAX_VALUE);
private static final String logDirectory = TestConstant.BASE_OUTPUT_PATH.concat("wal-test");
- private static final String devicePath = "root.test_sg.test_d";
+ private static final String databasePath = "root.test_sg";
+ private static final String devicePath = databasePath + ".test_d";
+ private static final String dataRegionId = "1";
private static final WALInsertNodeCache cache = WALInsertNodeCache.getInstance();
private WALMode prevMode;
private boolean prevIsClusterMode;
@@ -74,7 +76,7 @@ public class WALInsertNodeCacheTest {
@Test
public void testLoadUnsealedWALFile() throws Exception {
- IMemTable memTable = new PrimitiveMemTable();
+ IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode.onMemTableCreated(memTable, logDirectory + "/" + "fake.tsfile");
InsertRowNode node1 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(1);
@@ -91,7 +93,7 @@ public class WALInsertNodeCacheTest {
@Test
public void testBatchLoad() throws Exception {
// write memTable1
- IMemTable memTable1 = new PrimitiveMemTable();
+ IMemTable memTable1 = new PrimitiveMemTable(databasePath, dataRegionId);
walNode.onMemTableCreated(memTable1, logDirectory + "/" + "fake1.tsfile");
InsertRowNode node1 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(1);
@@ -102,7 +104,7 @@ public class WALInsertNodeCacheTest {
WALFlushListener flushListener2 = walNode.log(memTable1.getMemTableId(), node2);
WALEntryPosition position2 = flushListener2.getWalEntryHandler().getWalEntryPosition();
// write memTable2
- IMemTable memTable2 = new PrimitiveMemTable();
+ IMemTable memTable2 = new PrimitiveMemTable(databasePath, dataRegionId);
walNode.onMemTableCreated(memTable2, logDirectory + "/" + "fake2.tsfile");
InsertRowNode node3 = getInsertRowNode(devicePath, System.currentTimeMillis());
node1.setSearchIndex(3);
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index b6e5fbee5e8..510617eafd2 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -36,6 +36,7 @@ public enum Metric {
SERIES_SLOT_NUM_IN_DATABASE("series_slot_num_in_database"),
TIME_SLOT_NUM_IN_DATABASE("time_slot_num_in_database"),
REGION_GROUP_NUM_IN_DATABASE("region_group_num_in_database"),
+ REPLICATION_FACTOR("replication_factor"),
// protocol related
ENTRY("entry"),
SESSION_IDLE_TIME("session_idle_time"),