[Deferred Flushing] Implementation of deferred flushing strategy. (#14724)
Co-authored-by: DeferredFlushing <>
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index a9ec1ee..f3d0c7b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -410,6 +410,9 @@
/** The sort algorithm used in TVList */
private TVListSortAlgorithm tvListSortAlgorithm = TVListSortAlgorithm.TIM;
+ /** The value of primitive topk area size in the tvlist */
+ private int seqMemtableTopKSize = 0;
+
/** When average series point number reaches this, flush the memtable to disk */
private int avgSeriesPointNumberThreshold = 100000;
@@ -2070,6 +2073,14 @@
this.tvListSortAlgorithm = tvListSortAlgorithm;
}
+ public int getSeqMemtableTopKSize() {
+ return seqMemtableTopKSize;
+ }
+
+ public void setSeqMemtableTopKSize(int seqMemtableTopKSize) {
+ this.seqMemtableTopKSize = seqMemtableTopKSize;
+ }
+
public int getAvgSeriesPointNumberThreshold() {
return avgSeriesPointNumberThreshold;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 9160cb8..2b728ca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -387,6 +387,11 @@
properties.getProperty(
"tvlist_sort_algorithm", conf.getTvListSortAlgorithm().toString())));
+ conf.setSeqMemtableTopKSize(
+ (Integer.parseInt(
+ properties.getProperty(
+ "seq_memtable_topk_size", Integer.toString(conf.getSeqMemtableTopKSize())))));
+
conf.setAvgSeriesPointNumberThreshold(
Integer.parseInt(
properties.getProperty(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 20cbfef..eff6ddb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -1391,7 +1391,6 @@
version,
0,
0);
-
return getTsFileProcessor(sequence, filePath, timePartitionId);
}
@@ -1503,6 +1502,7 @@
if (!workSequenceTsFileProcessors.containsKey(tsFileProcessor.getTimeRangeId())) {
timePartitionIdVersionControllerMap.remove(tsFileProcessor.getTimeRangeId());
}
+ logger.info("close an unsequence tsfile processor {}", databaseName + "-" + dataRegionId);
}
if (workSequenceTsFileProcessors.get(tsFileProcessor.getTimeRangeId()) == null
&& workUnsequenceTsFileProcessors.get(tsFileProcessor.getTimeRangeId()) == null) {
@@ -1663,7 +1663,8 @@
for (TsFileProcessor tsFileProcessor : tsFileProcessors) {
if (tsFileProcessor.getWorkMemTableUpdateTime() < timeLowerBound) {
logger.info(
- "Exceed sequence memtable flush interval, so flush working memtable of time partition {} in database {}[{}]",
+ "Exceed sequence memtable {} flush interval, so flush working memtable of time partition {} in database {}[{}]",
+ tsFileProcessor.getWorkMemTable().getMemTableId(),
tsFileProcessor.getTimeRangeId(),
databaseName,
dataRegionId);
@@ -1689,7 +1690,8 @@
for (TsFileProcessor tsFileProcessor : tsFileProcessors) {
if (tsFileProcessor.getWorkMemTableUpdateTime() < timeLowerBound) {
logger.info(
- "Exceed unsequence memtable flush interval, so flush working memtable of time partition {} in database {}[{}]",
+ "Exceed unsequence memtable {} flush interval, so flush working memtable of time partition {} in database {}[{}]",
+ tsFileProcessor.getWorkMemTable().getMemTableId(),
tsFileProcessor.getTimeRangeId(),
databaseName,
dataRegionId);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
index ab5440d..9becb3c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java
@@ -110,6 +110,13 @@
memTable.getSeriesNumber() == 0
? 0
: memTable.getTotalPointsNum() / memTable.getSeriesNumber();
+ LOGGER.info(
+ "The memTable {} size of SG {} is {}, the avg series points num in chunk is {}, total timeseries number is {}",
+ memTable.getMemTableId(),
+ storageGroup,
+ memTable.memSize(),
+ avgSeriesPointsNum,
+ memTable.getSeriesNumber());
WRITING_METRICS.recordFlushingMemTableStatus(
storageGroup,
memTable.memSize(),
@@ -205,6 +212,13 @@
MetricLevel.CORE,
Tag.NAME.toString(),
"flush");
+
+ LOGGER.info(
+ "Database {} flushing memtable {} with {} has finished! Time consumption: {}ms",
+ storageGroup,
+ memTable.getMemTableId(),
+ memTable,
+ System.currentTimeMillis() - start);
}
/** encoding task (second task of pipeline) */
@@ -274,6 +288,17 @@
databaseName ->
recordFlushPointsMetricInternal(
memTable.getTotalPointsNum(), databaseName, dataRegionId));
+
+ LOGGER.info(
+ "Database {}, flushing memtable {} with size {} and points {} in file "
+ + "{} into disk: Encoding data cost "
+ + "{} ms.",
+ storageGroup,
+ memTable.getMemTableId(),
+ memTable.memSize(),
+ memTable.getTotalPointsNum(),
+ writer.getFile().getName(),
+ memSerializeTime);
WRITING_METRICS.recordFlushCost(WritingMetrics.FLUSH_STAGE_ENCODING, memSerializeTime);
}
};
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 e54e7ef..0bdebb2 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
@@ -37,6 +37,7 @@
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils;
import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -88,9 +89,9 @@
private long minPlanIndex = Long.MAX_VALUE;
- private final long memTableId = memTableIdCounter.incrementAndGet();
+ private long memTableId = memTableIdCounter.incrementAndGet();
- private final long createdTime = System.currentTimeMillis();
+ private long createdTime = System.currentTimeMillis();
/** this time is updated by the timed flush, same as createdTime when the feature is disabled. */
private long updateTime = createdTime;
@@ -126,6 +127,34 @@
this.memTableMap = memTableMap;
}
+ /** create memtable from last memtable, preserve topk points */
+ protected AbstractMemTable(
+ String database,
+ String dataRegionId,
+ Map<IDeviceID, IWritableMemChunkGroup> memTableMap,
+ AbstractMemTable lastMemTable) {
+ this.database = database;
+ this.dataRegionId = dataRegionId;
+ this.memTableMap = memTableMap;
+ for (IDeviceID key : memTableMap.keySet()) {
+ Map<String, IWritableMemChunk> memChunkMap = memTableMap.get(key).getMemChunkMap();
+ for (String schema : memChunkMap.keySet()) {
+ IWritableMemChunk memChunk = memChunkMap.get(schema);
+ totalPointsNum += memChunk.getTVList().rowCount();
+ memSize +=
+ (long) memChunk.getTVList().rowCount()
+ * memChunk.getSchema().getType().getDataTypeSize();
+ }
+ }
+ totalPointsNumThreshold = lastMemTable.getTotalPointsNumThreshold();
+ createdTime = System.currentTimeMillis();
+ memTableId = memTableIdCounter.incrementAndGet();
+ seriesNumber = lastMemTable.getSeriesNumber();
+ tvListRamCost = lastMemTable.getTVListsRamCost() * totalPointsNum / lastMemTable.totalPointsNum;
+ shouldFlush = false;
+ flushStatus = FlushStatus.WORKING;
+ }
+
@Override
public Map<IDeviceID, IWritableMemChunkGroup> getMemTableMap() {
return memTableMap;
@@ -440,6 +469,16 @@
}
@Override
+ public void setMemSize(long m) {
+ memSize = memSize + m;
+ }
+
+ @Override
+ public void setTotalPointsNum(long m) {
+ totalPointsNum = totalPointsNum + m;
+ }
+
+ @Override
public boolean reachTotalPointNumThreshold() {
if (totalPointsNum == 0) {
return false;
@@ -530,6 +569,21 @@
}
@Override
+ public IMemTable divide() throws TopkDivideMemoryNotEnoughException {
+ Map<IDeviceID, IWritableMemChunkGroup> topkMemTableMap = new HashMap<>();
+ for (IDeviceID key : memTableMap.keySet()) {
+ IWritableMemChunkGroup chunkGroupTemp = memTableMap.get(key).divide();
+ topkMemTableMap.put(key, chunkGroupTemp);
+ }
+ PrimitiveMemTable topkMemtable =
+ new PrimitiveMemTable(database, dataRegionId, topkMemTableMap, this);
+ memSize -= topkMemtable.memSize();
+ totalPointsNum -= topkMemtable.getTotalPointsNum();
+ tvListRamCost -= topkMemtable.getTVListsRamCost();
+ return topkMemtable;
+ }
+
+ @Override
public void addTVListRamCost(long cost) {
this.tvListRamCost += cost;
}
@@ -689,6 +743,15 @@
return latestTimeForEachDevice;
}
+ @Override
+ public Map<String, Long> getTopKTime() {
+ Map<String, Long> latestTimeForEachDevice = new HashMap<>();
+ for (Entry<IDeviceID, IWritableMemChunkGroup> entry : memTableMap.entrySet()) {
+ latestTimeForEachDevice.put(entry.getKey().toStringID(), entry.getValue().getTopKTime());
+ }
+ return latestTimeForEachDevice;
+ }
+
public static class Factory {
private Factory() {
// Empty constructor
@@ -734,4 +797,8 @@
public boolean isTotallyGeneratedByPipe() {
return this.isTotallyGeneratedByPipe.get();
}
+
+ public long getTotalPointsNumThreshold() {
+ return totalPointsNumThreshold;
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
index 64a56ee..55ad484 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
@@ -25,6 +25,7 @@
import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -254,11 +255,19 @@
}
@Override
+ public void setSchema(IMeasurementSchema s) {}
+
+ @Override
public long getMaxTime() {
return list.getMaxTime();
}
@Override
+ public long getTopKTime() {
+ return list.getTopKTime();
+ }
+
+ @Override
public synchronized TVList getSortedTvListForQuery() {
sortTVList();
// increase reference count
@@ -496,6 +505,11 @@
}
@Override
+ public IWritableMemChunk divide() throws TopkDivideMemoryNotEnoughException {
+ return new AlignedWritableMemChunk(schemaList, (AlignedTVList) list.divide());
+ }
+
+ @Override
public boolean isEmpty() {
return list.rowCount() == 0;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
index 17b12a7..875c113 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
@@ -23,6 +23,7 @@
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternUtil;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.utils.BitMap;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
@@ -143,6 +144,18 @@
return memChunk.getMaxTime();
}
+ @Override
+ public long getTopKTime() {
+ return memChunk.getTopKTime();
+ }
+
+ @Override
+ public IWritableMemChunkGroup divide() throws TopkDivideMemoryNotEnoughException {
+ AlignedWritableMemChunkGroup topkMemChunkGroup = new AlignedWritableMemChunkGroup();
+ topkMemChunkGroup.memChunk = (AlignedWritableMemChunk) memChunk.divide();
+ return topkMemChunkGroup;
+ }
+
public AlignedWritableMemChunk getAlignedMemChunk() {
return memChunk;
}
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 8e1a883..fc9a542 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
@@ -28,6 +28,7 @@
import org.apache.iotdb.db.storageengine.dataregion.flush.FlushStatus;
import org.apache.iotdb.db.storageengine.dataregion.modification.Modification;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
@@ -65,6 +66,10 @@
/** @return memory usage */
long memSize();
+ void setMemSize(long m);
+
+ void setTotalPointsNum(long m);
+
/** only used when mem control enabled */
void addTVListRamCost(long cost);
@@ -132,6 +137,8 @@
*/
void delete(PartialPath path, PartialPath devicePath, long startTimestamp, long endTimestamp);
+ /** divide the memtable into the topk and regular ones @ return the topk memtable. */
+ IMemTable divide() throws TopkDivideMemoryNotEnoughException;
/**
* Make a copy of this MemTable.
*
@@ -185,4 +192,6 @@
void markAsNotGeneratedByPipe();
boolean isTotallyGeneratedByPipe();
+
+ Map<String, Long> getTopKTime();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
index 4c07c3e..4b643cb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunk.java
@@ -20,6 +20,7 @@
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue;
import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.BitMap;
@@ -83,6 +84,8 @@
IMeasurementSchema getSchema();
+ void setSchema(IMeasurementSchema s);
+
/**
* served for read requests.
*
@@ -128,6 +131,10 @@
return Long.MAX_VALUE;
}
+ default long getTopKTime() {
+ return Long.MAX_VALUE;
+ }
+
/** @return how many points are deleted */
int delete(long lowerBound, long upperBound);
@@ -141,5 +148,7 @@
long getLastPoint();
+ IWritableMemChunk divide() throws TopkDivideMemoryNotEnoughException;
+
boolean isEmpty();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
index ecaf8b2..eb3c1f0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
@@ -21,6 +21,7 @@
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.utils.BitMap;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
@@ -53,5 +54,9 @@
long getCurrentTVListSize(String measurement);
+ IWritableMemChunkGroup divide() throws TopkDivideMemoryNotEnoughException;
+
long getMaxTime();
+
+ long getTopKTime();
}
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 70b0f90..cfdd9c2 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
@@ -37,6 +37,14 @@
super(database, dataRegionId, memTableMap);
}
+ public PrimitiveMemTable(
+ String database,
+ String dataRegionId,
+ Map<IDeviceID, IWritableMemChunkGroup> memTableMap,
+ AbstractMemTable lastMemTable) {
+ super(database, dataRegionId, memTableMap, lastMemTable);
+ }
+
@Override
public IMemTable copy() {
Map<IDeviceID, IWritableMemChunkGroup> newMap = new HashMap<>(getMemTableMap());
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 953606e..fd8331b7 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
@@ -62,6 +62,7 @@
import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
@@ -93,6 +94,7 @@
import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.GET_QUERY_RESOURCE_FROM_MEM;
import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.FLUSHING_MEMTABLE;
import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.WORKING_MEMTABLE;
+import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.MEMTABLE_TOPK_SIZE;
@SuppressWarnings("java:S1135") // ignore todos
public class TsFileProcessor {
@@ -180,6 +182,8 @@
private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS =
PerformanceOverviewMetrics.getInstance();
+ private File tsfileTemp;
+
@SuppressWarnings("squid:S107")
public TsFileProcessor(
String storageGroupName,
@@ -324,6 +328,39 @@
dataRegionInfo.getDataRegion().getDatabaseName(),
dataRegionInfo.getDataRegion().getDataRegionId());
walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath());
+ logger.info(
+ "Succeed to create {} memTable {}",
+ sequence ? "Sequence" : "Unsequence",
+ workMemTable.getMemTableId());
+ }
+
+ private void createNewWorkingMemTable(IMemTable tobeFlushed) throws WriteProcessException {
+ if (MEMTABLE_TOPK_SIZE != 0) {
+ try {
+ logger.info(
+ "Try to divide memTable {} with size {} and points {}",
+ tobeFlushed.getMemTableId(),
+ tobeFlushed.memSize(),
+ tobeFlushed.getTotalPointsNum());
+ workMemTable = tobeFlushed.divide();
+ walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath());
+ logger.info(
+ "Succeed to divide {} memTable {} with size {} and points {} to "
+ + "new workMemTable {} with size {} and points {} to",
+ sequence ? "Sequence" : "Unsequence",
+ tobeFlushed.getMemTableId(),
+ tobeFlushed.memSize(),
+ tobeFlushed.getTotalPointsNum(),
+ workMemTable.getMemTableId(),
+ workMemTable.memSize(),
+ workMemTable.getTotalPointsNum());
+ } catch (TopkDivideMemoryNotEnoughException e) {
+ logger.warn(e.getMessage());
+ createNewWorkingMemTable();
+ }
+ } else {
+ createNewWorkingMemTable();
+ }
}
/**
@@ -854,18 +891,19 @@
try {
if (logger.isDebugEnabled()) {
if (workMemTable != null) {
- logger.debug(
- "{}: flush a working memtable in async close tsfile {}, memtable size: {}, tsfile "
- + "size: {}, plan index: [{}, {}], progress index: {}",
+ logger.info(
+ "{}: flush working memtable {} with size: {}, tsfile "
+ + "size: {}, plan index: [{}, {}], progress index: {} in async close tsfile {}",
storageGroupName,
- tsFileResource.getTsFile().getAbsolutePath(),
+ workMemTable.getMemTableId(),
workMemTable.memSize(),
tsFileResource.getTsFileSize(),
workMemTable.getMinPlanIndex(),
workMemTable.getMaxPlanIndex(),
- tsFileResource.getMaxProgressIndex());
+ tsFileResource.getMaxProgressIndex(),
+ tsFileResource.getTsFile().getAbsolutePath());
} else {
- logger.debug(
+ logger.info(
"{}: flush a NotifyFlushMemTable in async close tsfile {}, tsfile size: {}",
storageGroupName,
tsFileResource.getTsFile().getAbsolutePath(),
@@ -899,6 +937,12 @@
// When invoke closing TsFile after insert data to memTable, we shouldn't flush until invoke
// flushing memTable in System module.
Future<?> future = addAMemtableIntoFlushingList(tmpMemTable);
+ logger.info(
+ "Memtable {}: {} with size {} and points {} has been added to flushing list",
+ tmpMemTable.getMemTableId(),
+ tmpMemTable,
+ tmpMemTable.memSize(),
+ tmpMemTable.getTotalPointsNum());
shouldClose = true;
return future;
} catch (Exception e) {
@@ -982,7 +1026,9 @@
return;
}
logger.info(
- "Async flush a memtable to tsfile: {}", tsFileResource.getTsFile().getAbsolutePath());
+ "Async flush a memtable {} to tsfile: {}",
+ workMemTable.getMemTableId(),
+ tsFileResource.getTsFile().getAbsolutePath());
addAMemtableIntoFlushingList(workMemTable);
} catch (Exception e) {
logger.error(
@@ -1007,7 +1053,9 @@
private Future<?> addAMemtableIntoFlushingList(IMemTable tobeFlushed) throws IOException {
Map<String, Long> lastTimeForEachDevice = new HashMap<>();
if (sequence) {
- lastTimeForEachDevice = tobeFlushed.getMaxTime();
+ // TODO:修改lastFlushTime的取法
+ lastTimeForEachDevice = tobeFlushed.getTopKTime();
+ // lastTimeForEachDevice = tobeFlushed.getMaxTime();
// If some devices have been removed in MemTable, the number of device in MemTable and
// tsFileResource will not be the same. And the endTime of these devices in resource will be
// Long.minValue.
@@ -1027,7 +1075,18 @@
updateLatestFlushTimeCallback.call(this, lastTimeForEachDevice, lastWorkMemtableFlushTime);
SystemInfo.getInstance().addFlushingMemTableCost(tobeFlushed.getTVListsRamCost());
- flushingMemTables.addLast(tobeFlushed);
+ // TODO:如果是顺序区并且k>0,拆分之后分别进入不同的waitinglist中，否则的话仍然按照原方式刷盘
+ if (MEMTABLE_TOPK_SIZE == 0) {
+ flushingMemTables.addLast(tobeFlushed);
+ workMemTable = null;
+ } else {
+ try {
+ createNewWorkingMemTable(tobeFlushed);
+ } catch (WriteProcessException e) {
+ throw new RuntimeException(e);
+ }
+ flushingMemTables.addLast(tobeFlushed);
+ }
if (logger.isDebugEnabled()) {
logger.debug(
"{}: {} Memtable (signal = {}) is added into the flushing Memtable, queue size = {}",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
index c1a71d3..c750fb6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
@@ -24,6 +24,7 @@
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -55,7 +56,13 @@
this.list = TVList.newList(schema.getType());
}
- private WritableMemChunk() {}
+ public WritableMemChunk(IMeasurementSchema schema, TVList ls) {
+ this.schema = schema;
+ this.list = ls;
+ }
+
+ // private WritableMemChunk() {}
+ public WritableMemChunk() {}
@Override
public boolean writeWithFlushCheck(long insertTime, Object objectValue) {
@@ -252,11 +259,21 @@
}
@Override
+ public void setSchema(IMeasurementSchema s) {
+ this.schema = s;
+ }
+
+ @Override
public long getMaxTime() {
return list.getMaxTime();
}
@Override
+ public long getTopKTime() {
+ return list.getTopKTime();
+ }
+
+ @Override
public long getFirstPoint() {
if (list.rowCount() == 0) {
return Long.MAX_VALUE;
@@ -275,6 +292,11 @@
}
@Override
+ public IWritableMemChunk divide() throws TopkDivideMemoryNotEnoughException {
+ return new WritableMemChunk(this.schema, list.divide());
+ }
+
+ @Override
public boolean isEmpty() {
return list.rowCount() == 0;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
index 707eace..dbe03f1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
@@ -23,6 +23,7 @@
import org.apache.iotdb.commons.path.PathPatternUtil;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils;
+import org.apache.iotdb.db.utils.datastructure.TopkDivideMemoryNotEnoughException;
import org.apache.iotdb.tsfile.utils.BitMap;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
@@ -173,6 +174,25 @@
}
@Override
+ public long getTopKTime() {
+ long maxTime = Long.MIN_VALUE;
+ for (IWritableMemChunk memChunk : memChunkMap.values()) {
+ maxTime = Math.max(maxTime, memChunk.getTopKTime());
+ }
+ return maxTime;
+ }
+
+ @Override
+ public IWritableMemChunkGroup divide() throws TopkDivideMemoryNotEnoughException {
+ WritableMemChunkGroup topkMemChunkGroup = new WritableMemChunkGroup();
+ for (String key : memChunkMap.keySet()) {
+ IWritableMemChunk topkMemChunk = memChunkMap.get(key).divide();
+ topkMemChunkGroup.memChunkMap.put(key, topkMemChunk);
+ }
+ return topkMemChunkGroup;
+ }
+
+ @Override
public int serializedSize() {
int size = 0;
size += Integer.BYTES;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java
index 33264fc..4f1edd8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/PrimitiveArrayManager.java
@@ -40,6 +40,7 @@
private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
public static final int ARRAY_SIZE = CONFIG.getPrimitiveArraySize();
+ public static final int MEMTABLE_TOPK_SIZE = CONFIG.getSeqMemtableTopKSize();
public static final TVListSortAlgorithm TVLIST_SORT_ALGORITHM = CONFIG.getTvListSortAlgorithm();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
index 4092db7..ee451ac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
@@ -43,8 +43,7 @@
import java.util.List;
import java.util.Objects;
-import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
-import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.TVLIST_SORT_ALGORITHM;
+import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.*;
import static org.apache.iotdb.db.utils.MemUtils.getBinarySize;
import static org.apache.iotdb.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER;
import static org.apache.iotdb.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
@@ -647,6 +646,37 @@
}
}
+ @Override
+ public TVList divide() throws TopkDivideMemoryNotEnoughException {
+ if (MEMTABLE_TOPK_SIZE > rowCount || MEMTABLE_TOPK_SIZE < ARRAY_SIZE) {
+ throw new TopkDivideMemoryNotEnoughException(
+ String.format(
+ "WARNING: MEMTABLE_TOPK_SIZE is bigger than%d "
+ + "the TVList's row count %d"
+ + "or is smaller than ARRAY_SIZE %d",
+ rowCount, MEMTABLE_TOPK_SIZE, ARRAY_SIZE));
+ }
+ AlignedTVList topkTVList = AlignedTVList.newAlignedList(dataTypes);
+ int truncatedIndex = (rowCount - MEMTABLE_TOPK_SIZE + ARRAY_SIZE - 1) / ARRAY_SIZE;
+
+ for (int i = truncatedIndex + 1; i <= rowCount / ARRAY_SIZE; i++) {
+ topkTVList.timestamps.add(timestamps.get(i));
+ for (int colIndex = 0; colIndex < values.size(); colIndex++) {
+ topkTVList.values.get(colIndex).add(values.get(colIndex).get(i));
+ topkTVList.bitMaps.get(colIndex).add(bitMaps.get(colIndex).get(i));
+ }
+ }
+ for (int i = rowCount / ARRAY_SIZE; i >= truncatedIndex + 1; i--) {
+ timestamps.remove(timestamps.size() - 1);
+ for (int colIndex = 0; colIndex < values.size(); colIndex++) {
+ topkTVList.values.get(colIndex).remove(values.get(colIndex).size() - 1);
+ topkTVList.bitMaps.get(colIndex).remove(bitMaps.get(colIndex).size() - 1);
+ }
+ }
+ rowCount = truncatedIndex * ARRAY_SIZE;
+ topKTime = timestamps.get(truncatedIndex - 1)[ARRAY_SIZE - 1];
+ return topkTVList;
+ }
/**
* Get the row index value in index column.
*
@@ -749,6 +779,8 @@
checkExpansion();
}
}
+ sort(Math.max(0, start - MEMTABLE_TOPK_SIZE), end);
+ topKTime = Math.max(topKTime, getTime(Math.max(0, end - MEMTABLE_TOPK_SIZE)));
}
private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex, int remaining) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackwardSort.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackwardSort.java
index 6271a31..21783bc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackwardSort.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackwardSort.java
@@ -53,7 +53,7 @@
/**
* check block-inversions to find the proper block_size, which is a multiple of array_size. For
- * totally ordered, the block_size will equals to array_size For totally reverse ordered, the
+ * totally ordered, the block_size will equal to array_size For totally reverse ordered, the
* block_size will equals to the rowCount. INVERSION_RATIOS_THRESHOLD=0.005 is a empiric value.
*
* @param timestamps
@@ -132,6 +132,6 @@
* @param hi
*/
default void sortBlock(int lo, int hi) {
- qsort(lo, hi);
+ qSort(lo, hi);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
index c64dbc9..a59fa4f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
@@ -34,8 +34,7 @@
import java.util.ArrayList;
import java.util.List;
-import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
-import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.TVLIST_SORT_ALGORITHM;
+import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.*;
public abstract class IntTVList extends TVList {
// list of primitive array, add 1 when expanded -> int primitive array
@@ -77,14 +76,59 @@
@Override
public void putInt(long timestamp, int value) {
checkExpansion();
- int arrayIndex = rowCount / ARRAY_SIZE;
- int elementIndex = rowCount % ARRAY_SIZE;
- maxTime = Math.max(maxTime, timestamp);
- timestamps.get(arrayIndex)[elementIndex] = timestamp;
- values.get(arrayIndex)[elementIndex] = value;
- rowCount++;
- if (sorted && rowCount > 1 && timestamp < getTime(rowCount - 2)) {
- sorted = false;
+ if (MEMTABLE_TOPK_SIZE == 0) {
+ int arrayIndex = rowCount / ARRAY_SIZE;
+ int elementIndex = rowCount % ARRAY_SIZE;
+ maxTime = Math.max(maxTime, timestamp);
+ topKTime = Math.max(topKTime, timestamp);
+ timestamps.get(arrayIndex)[elementIndex] = timestamp;
+ values.get(arrayIndex)[elementIndex] = value;
+ rowCount++;
+ if (sorted && rowCount > 1 && timestamp < getTime(rowCount - 2)) {
+ sorted = false;
+ }
+ } else {
+ int arrayIndex;
+ int elementIndex;
+ int waitlen = Math.min(rowCount, MEMTABLE_TOPK_SIZE);
+ int tempRowCount = rowCount - 1; // 记录向前比较的截止位置
+ while (waitlen > 0) {
+ waitlen--;
+ arrayIndex = tempRowCount / ARRAY_SIZE;
+ elementIndex = tempRowCount % ARRAY_SIZE;
+ if (timestamps.get(arrayIndex)[elementIndex] > timestamp) {
+ tempRowCount--;
+ continue;
+ }
+ break;
+ }
+ arrayIndex = rowCount / ARRAY_SIZE;
+ elementIndex = rowCount % ARRAY_SIZE;
+ int arrayIndexLeft = (rowCount - 1) / ARRAY_SIZE;
+ int elementIndexLeft = (rowCount - 1) % ARRAY_SIZE;
+ for (int i = rowCount - 1; i > tempRowCount; i--) {
+ timestamps.get(arrayIndex)[elementIndex] = timestamps.get(arrayIndexLeft)[elementIndexLeft];
+ values.get(arrayIndex)[elementIndex] = values.get(arrayIndexLeft)[elementIndexLeft];
+ arrayIndex = arrayIndexLeft;
+ elementIndex = elementIndexLeft;
+ arrayIndexLeft = (i - 1) / ARRAY_SIZE;
+ elementIndexLeft = (i - 1) % ARRAY_SIZE;
+ sortCount++;
+ }
+ arrayIndex = (tempRowCount + 1) / ARRAY_SIZE;
+ elementIndex = (tempRowCount + 1) % ARRAY_SIZE;
+ maxTime = Math.max(maxTime, timestamp);
+ timestamps.get(arrayIndex)[elementIndex] = timestamp;
+ values.get(arrayIndex)[elementIndex] = value;
+ if (rowCount > MEMTABLE_TOPK_SIZE) {
+ arrayIndex = (rowCount - MEMTABLE_TOPK_SIZE - 1) / ARRAY_SIZE;
+ elementIndex = (rowCount - MEMTABLE_TOPK_SIZE - 1) % ARRAY_SIZE;
+ topKTime = Math.max(topKTime, timestamps.get(arrayIndex)[elementIndex]);
+ }
+ rowCount++;
+ if (sorted && rowCount > 1 && waitlen == 0) {
+ sorted = false;
+ }
}
}
@@ -124,6 +168,37 @@
}
@Override
+ public TVList divide() throws TopkDivideMemoryNotEnoughException {
+ if (MEMTABLE_TOPK_SIZE > rowCount || MEMTABLE_TOPK_SIZE < ARRAY_SIZE) {
+ throw new TopkDivideMemoryNotEnoughException(
+ String.format(
+ "WARNING: MEMTABLE_TOPK_SIZE %d is bigger than "
+ + "the TVList's row count %d, "
+ + "or is smaller than ARRAY_SIZE %d",
+ MEMTABLE_TOPK_SIZE, rowCount, ARRAY_SIZE));
+ }
+ IntTVList topkTVList = IntTVList.newList();
+ int truncatedIndex = rowCount - MEMTABLE_TOPK_SIZE;
+ int truncatedArrayIndex =
+ truncatedIndex / ARRAY_SIZE; // no matter truncatedIndex in or not in the block
+ truncatedIndex = truncatedArrayIndex * ARRAY_SIZE;
+ for (int i = truncatedArrayIndex; i < timestamps.size(); i++) {
+ topkTVList.timestamps.add(timestamps.get(i));
+ topkTVList.values.add(values.get(i));
+ }
+ for (int i = timestamps.size() - 1; i >= truncatedArrayIndex; i--) {
+ timestamps.remove(timestamps.size() - 1);
+ values.remove(values.size() - 1);
+ }
+ topkTVList.rowCount = rowCount - truncatedIndex;
+ topkTVList.sorted = true;
+ topkTVList.topKTime = topkTVList.getTime(0);
+ rowCount = truncatedIndex;
+ topKTime = topkTVList.getTopKTime();
+ return topkTVList;
+ }
+
+ @Override
public TimeValuePair getTimeValuePair(int index) {
return new TimeValuePair(
getTime(index), TsPrimitiveType.getByType(TSDataType.INT32, getInt(index)));
@@ -159,47 +234,8 @@
@Override
public void putInts(long[] time, int[] value, BitMap bitMap, int start, int end) {
- checkExpansion();
-
- int idx = start;
- // constraint: time.length + timeIdxOffset == value.length
- int timeIdxOffset = 0;
- if (bitMap != null && !bitMap.isAllUnmarked()) {
- // time array is a reference, should clone necessary time values
- long[] clonedTime = new long[end - start];
- System.arraycopy(time, start, clonedTime, 0, end - start);
- time = clonedTime;
- timeIdxOffset = start;
- // drop null at the end of value array
- int nullCnt =
- dropNullValThenUpdateMaxTimeAndSorted(time, value, bitMap, start, end, timeIdxOffset);
- end -= nullCnt;
- } else {
- updateMaxTimeAndSorted(time, start, end);
- }
-
- while (idx < end) {
- int inputRemaining = end - idx;
- int arrayIdx = rowCount / ARRAY_SIZE;
- int elementIdx = rowCount % ARRAY_SIZE;
- int internalRemaining = ARRAY_SIZE - elementIdx;
- if (internalRemaining >= inputRemaining) {
- // the remaining inputs can fit the last array, copy all remaining inputs into last array
- System.arraycopy(
- time, idx - timeIdxOffset, timestamps.get(arrayIdx), elementIdx, inputRemaining);
- System.arraycopy(value, idx, values.get(arrayIdx), elementIdx, inputRemaining);
- rowCount += inputRemaining;
- break;
- } else {
- // the remaining inputs cannot fit the last array, fill the last array and create a new
- // one and enter the next loop
- System.arraycopy(
- time, idx - timeIdxOffset, timestamps.get(arrayIdx), elementIdx, internalRemaining);
- System.arraycopy(value, idx, values.get(arrayIdx), elementIdx, internalRemaining);
- idx += internalRemaining;
- rowCount += internalRemaining;
- checkExpansion();
- }
+ for (int i = start; i < end; i++) {
+ putInt(time[i], value[i]);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
index 5e6a9de..d60067a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
@@ -30,12 +30,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
int srcV = getValueIndex(src);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBinaryTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBinaryTVList.java
index dcdcbc2..ee4e6f9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBinaryTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBinaryTVList.java
@@ -42,12 +42,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
Binary srcV = getBinary(src);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBooleanTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBooleanTVList.java
index e4fbfae..c7b4347 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBooleanTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickBooleanTVList.java
@@ -39,12 +39,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
boolean srcV = getBoolean(src);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickDoubleTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickDoubleTVList.java
index 852844f..584c8f8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickDoubleTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickDoubleTVList.java
@@ -39,12 +39,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
double srcV = getDouble(src);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickFloatTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickFloatTVList.java
index 409a00a..9048525 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickFloatTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickFloatTVList.java
@@ -39,12 +39,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
float srcV = getFloat(src);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickIntTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickIntTVList.java
index bc44cf7..d24842b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickIntTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickIntTVList.java
@@ -22,12 +22,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
public void swap(int p, int q) {
int valueP = getInt(p);
long timeP = getTime(p);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickLongTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickLongTVList.java
index 16d8fab..3d4f1ab 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickLongTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickLongTVList.java
@@ -39,12 +39,17 @@
@Override
public void sort() {
if (!sorted) {
- qsort(0, rowCount - 1);
+ qSort(0, rowCount - 1);
}
sorted = true;
}
@Override
+ public void sort(int lo, int hi) {
+ qSort(lo, hi);
+ }
+
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
long srcV = getLong(src);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickSort.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickSort.java
index 8229781..9bad94d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickSort.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickSort.java
@@ -58,7 +58,7 @@
// }
// }
- default void qsort(int lo, int hi) {
+ default void qSort(int lo, int hi) {
if (lo < hi) {
// TODO: use insertion sort in smaller array
// if(hi - lo <= 32) {
@@ -66,8 +66,8 @@
// }
// partition
int pivotIndex = partition(lo, hi);
- qsort(lo, pivotIndex - 1);
- qsort(pivotIndex + 1, hi);
+ qSort(lo, pivotIndex - 1);
+ qSort(pivotIndex + 1, hi);
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
index f2b2615..8430df0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -42,6 +42,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
+import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.MEMTABLE_TOPK_SIZE;
import static org.apache.iotdb.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER;
import static org.apache.iotdb.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
@@ -56,6 +57,8 @@
protected boolean sorted = true;
protected long maxTime;
+ protected long topKTime;
+ protected int sortCount;
// record reference count of this tv list
// currently this reference will only be increase because we can't know when to decrease it
protected AtomicInteger referenceCount;
@@ -64,7 +67,9 @@
protected TVList() {
timestamps = new ArrayList<>();
rowCount = 0;
+ sortCount = 0;
maxTime = Long.MIN_VALUE;
+ topKTime = Long.MIN_VALUE;
referenceCount = new AtomicInteger();
}
@@ -107,6 +112,8 @@
public abstract void sort();
+ public abstract void sort(int lo, int hi);
+
public void increaseReferenceCount() {
referenceCount.incrementAndGet();
}
@@ -212,6 +219,10 @@
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
}
+ public TVList divide() throws TopkDivideMemoryNotEnoughException {
+ throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
+ }
+
public Object getAlignedValue(int index) {
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
}
@@ -229,6 +240,10 @@
return maxTime;
}
+ public long getTopKTime() {
+ return topKTime;
+ }
+
public long getVersion() {
return version;
}
@@ -328,6 +343,7 @@
for (int i = start; i < end; i++) {
inPutMinTime = Math.min(inPutMinTime, time[i]);
maxTime = Math.max(maxTime, time[i]);
+ topKTime = Math.max(topKTime, time[Math.max(0, i - MEMTABLE_TOPK_SIZE)]);
if (inputSorted && i < length - 1 && time[i] > time[i + 1]) {
inputSorted = false;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
index 7c6fa7f..819b00e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
@@ -50,7 +50,7 @@
(int[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.INT32, rowCount);
}
if (!sorted) {
- sort(0, rowCount);
+ this.timSort(0, rowCount);
}
clearSortedValue();
clearSortedTime();
@@ -58,7 +58,12 @@
}
@Override
- public void tim_set(int src, int dest) {
+ public void sort(int lo, int hi) {
+ timSort(lo, hi);
+ }
+
+ @Override
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBinaryTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBinaryTVList.java
index 03760fd..fccb64a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBinaryTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBinaryTVList.java
@@ -44,14 +44,19 @@
sortedValues =
(Binary[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.TEXT, rowCount);
}
- sort(0, rowCount);
+ timSort(0, rowCount);
clearSortedValue();
clearSortedTime();
sorted = true;
}
@Override
- public void tim_set(int src, int dest) {
+ public void sort(int lo, int hi) {
+ timSort(lo, hi);
+ }
+
+ @Override
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBooleanTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBooleanTVList.java
index 22b5647..04d43f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBooleanTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimBooleanTVList.java
@@ -43,7 +43,7 @@
(boolean[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.BOOLEAN, rowCount);
}
if (!sorted) {
- sort(0, rowCount);
+ timSort(0, rowCount);
}
clearSortedValue();
clearSortedTime();
@@ -51,7 +51,12 @@
}
@Override
- public void tim_set(int src, int dest) {
+ public void sort(int lo, int hi) {
+ timSort(lo, hi);
+ }
+
+ @Override
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimDoubleTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimDoubleTVList.java
index 3c0aa82..543a4e2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimDoubleTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimDoubleTVList.java
@@ -43,7 +43,7 @@
(double[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.DOUBLE, rowCount);
}
if (!sorted) {
- sort(0, rowCount);
+ timSort(0, rowCount);
}
clearSortedValue();
clearSortedTime();
@@ -51,7 +51,12 @@
}
@Override
- public void tim_set(int src, int dest) {
+ public void sort(int lo, int hi) {
+ timSort(lo, hi);
+ }
+
+ @Override
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimFloatTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimFloatTVList.java
index 2b1fb44..d360959 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimFloatTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimFloatTVList.java
@@ -44,7 +44,7 @@
(float[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.FLOAT, rowCount);
}
if (!sorted) {
- sort(0, rowCount);
+ timSort(0, rowCount);
}
clearSortedValue();
clearSortedTime();
@@ -52,7 +52,12 @@
}
@Override
- public void tim_set(int src, int dest) {
+ public void sort(int lo, int hi) {
+ timSort(lo, hi);
+ }
+
+ @Override
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimIntTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimIntTVList.java
index b3ae939..d2f153e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimIntTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimIntTVList.java
@@ -33,18 +33,22 @@
@Override
public void sort() {
+ sort(0, rowCount);
+ }
+
+ @Override
+ public void sort(int lo, int hi) {
+ int len = hi - lo;
if (sortedTimestamps == null
- || sortedTimestamps.length < PrimitiveArrayManager.getArrayRowCount(rowCount)) {
+ || sortedTimestamps.length < PrimitiveArrayManager.getArrayRowCount(len)) {
sortedTimestamps =
- (long[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.INT64, rowCount);
+ (long[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.INT64, len);
}
- if (sortedValues == null
- || sortedValues.length < PrimitiveArrayManager.getArrayRowCount(rowCount)) {
- sortedValues =
- (int[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.INT32, rowCount);
+ if (sortedValues == null || sortedValues.length < PrimitiveArrayManager.getArrayRowCount(len)) {
+ sortedValues = (int[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.INT32, len);
}
if (!sorted) {
- sort(0, rowCount);
+ timSort(0, len);
}
clearSortedValue();
clearSortedTime();
@@ -52,7 +56,7 @@
}
@Override
- public void tim_set(int src, int dest) {
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimLongTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimLongTVList.java
index e3a02e4..9096cd4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimLongTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimLongTVList.java
@@ -43,7 +43,7 @@
(long[][]) PrimitiveArrayManager.createDataListsByType(TSDataType.INT64, rowCount);
}
if (!sorted) {
- sort(0, rowCount);
+ timSort(0, rowCount);
}
clearSortedValue();
clearSortedTime();
@@ -51,7 +51,12 @@
}
@Override
- public void tim_set(int src, int dest) {
+ public void sort(int lo, int hi) {
+ timSort(lo, hi);
+ }
+
+ @Override
+ public void timSet(int src, int dest) {
set(src, dest);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimSort.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimSort.java
index 6358442..0426fda 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimSort.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimSort.java
@@ -27,7 +27,7 @@
int SMALL_ARRAY_LENGTH = 32;
/** the same as the 'set' function in TVList, the reason is to avoid two equal functions. */
- void tim_set(int src, int dest);
+ void timSet(int src, int dest);
void setFromSorted(int src, int dest);
@@ -55,7 +55,7 @@
* the entrance of tim_sort; 1. array_size <= 32, use binary sort. 2. recursively invoke merge
* sort.
*/
- default void sort(int lo, int hi) {
+ default void timSort(int lo, int hi) {
if (lo == hi) {
return;
}
@@ -65,8 +65,8 @@
return;
}
int mid = (lo + hi) >>> 1;
- sort(lo, mid);
- sort(mid, hi);
+ timSort(lo, mid);
+ timSort(mid, hi);
merge(lo, mid, hi);
}
@@ -127,7 +127,7 @@
*/
int n = start - left; // The number of elements to move
for (int i = n; i >= 1; i--) {
- tim_set(left + i - 1, left + i);
+ timSet(left + i - 1, left + i);
}
setPivotTo(left);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TopkDivideMemoryNotEnoughException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TopkDivideMemoryNotEnoughException.java
new file mode 100644
index 0000000..e1d8d72
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TopkDivideMemoryNotEnoughException.java
@@ -0,0 +1,8 @@
+package org.apache.iotdb.db.utils.datastructure;
+
+public class TopkDivideMemoryNotEnoughException extends Exception {
+
+ public TopkDivideMemoryNotEnoughException(String message) {
+ super(message);
+ }
+}
diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
index eb95972..c9d3a67 100644
--- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
+++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties
@@ -464,7 +464,7 @@
# Add a switch to enable separate sequence and unsequence data.
# If it is true, then data will be separated into seq and unseq data dir. If it is false, then all data will be written into unseq data dir.
# Datatype: boolean
-# enable_separate_data=true
+enable_separate_data=false
# What will the system do when unrecoverable error occurs.
# Datatype: String
@@ -509,9 +509,14 @@
# BACKWARD: backward sort
# tvlist_sort_algorithm=TIM
+# The number of points stay in the next tvlist
+# The default waiting size is 0
+# Datatype: int
+seq_memtable_topk_size=0
+
# When the average point number of timeseries in memtable exceeds this, the memtable is flushed to disk. The default threshold is 100000.
# Datatype: int
-# avg_series_point_number_threshold=100000
+avg_series_point_number_threshold=100000
# How many threads can concurrently flush. When <= 0, use CPU core number.
# Datatype: int
@@ -544,15 +549,15 @@
####################
# sequence space compaction: only compact the sequence files
# Datatype: boolean
-# enable_seq_space_compaction=true
+enable_seq_space_compaction=false
# unsequence space compaction: only compact the unsequence files
# Datatype: boolean
-# enable_unseq_space_compaction=true
+enable_unseq_space_compaction=false
# cross space compaction: compact the unsequence files into the overlapped sequence files
# Datatype: boolean
-# enable_cross_space_compaction=true
+enable_cross_space_compaction=false
# the selector of cross space compaction task
# Options: rewrite
@@ -644,7 +649,7 @@
# The interval of compaction task schedule
# Datatype: long, Unit: ms
-# compaction_schedule_interval_in_ms=60000
+compaction_schedule_interval_in_ms=600000000
# The interval of compaction task submission
# Datatype: long, Unit: ms
