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 11:57:23 UTC
[iotdb] branch rel/1.2 updated: Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113) (#11154)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new 39e2cf896a7 Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113) (#11154)
39e2cf896a7 is described below
commit 39e2cf896a70ac9f0608609a06fc0c33919fe5ba
Author: Potato <ta...@apache.org>
AuthorDate: Fri Sep 15 19:57:17 2023 +0800
Add replication factor metrics and add database tag in throughput metrics and schema metrics (#11113) (#11154)
Co-authored-by: ZhangHongYin <46...@users.noreply.github.com>
---
.../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"),