You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ea...@apache.org on 2019/12/12 12:48:17 UTC
[incubator-iotdb] branch nvmlogging created (now 5abe06a)
This is an automated email from the ASF dual-hosted git repository.
east pushed a change to branch nvmlogging
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 5abe06a update
This branch includes the following new commits:
new 5abe06a update
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: update
Posted by ea...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
east pushed a commit to branch nvmlogging
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 5abe06a4c0dbdde0fb10cd0827d7cba7c095015e
Author: mdf369 <95...@qq.com>
AuthorDate: Thu Dec 12 20:47:46 2019 +0800
update
---
.../iotdb/db/engine/flush/MemTableFlushTask.java | 3 +-
.../iotdb/db/engine/memtable/AbstractMemTable.java | 13 +-
.../db/engine/memtable/IWritableMemChunk.java | 6 +-
.../db/engine/memtable/PrimitiveMemTable.java | 21 +-
.../db/engine/storagegroup/TsFileProcessor.java | 22 +-
.../datastructure/AbstractTVList.java} | 151 ++--------
.../datastructure/NVMIntTVList.java} | 119 ++++----
.../iotdb/db/nvm/datastructure/NVMTVList.java | 123 ++++++++
.../apache/iotdb/db/nvm/executor/INVMExecutor.java | 12 +
.../apache/iotdb/db/nvm/executor/NVMExecutor.java | 134 +++++++++
.../db/nvm/memtable/NVMPrimitiveMemTable.java | 65 +++++
.../iotdb/db/nvm/memtable/NVMWritableMemChunk.java | 269 +++++++++++++++++
.../iotdb/db/nvm/rescon/NVMPrimitiveArrayPool.java | 68 +++++
.../iotdb/db/nvm/space/INVMSpaceManager.java | 9 +
.../apache/iotdb/db/nvm/space/NVMSpaceManager.java | 174 +++++++++++
.../org/apache/iotdb/db/rescon/MemTablePool.java | 141 ++++++---
.../apache/iotdb/db/rescon/TVListAllocator.java | 81 ++++--
.../iotdb/db/rescon/TVListAllocatorMBean.java | 2 +-
.../iotdb/db/utils/datastructure/BinaryTVList.java | 8 +-
.../db/utils/datastructure/BooleanTVList.java | 8 +-
.../iotdb/db/utils/datastructure/DoubleTVList.java | 8 +-
.../iotdb/db/utils/datastructure/FloatTVList.java | 8 +-
.../iotdb/db/utils/datastructure/IntTVList.java | 8 +-
.../iotdb/db/utils/datastructure/LongTVList.java | 8 +-
.../iotdb/db/utils/datastructure/TVList.java | 324 ++-------------------
.../iotdb/db/engine/memtable/MemTablePoolTest.java | 4 +-
.../iotdb/db/engine/storagegroup/PerfTest.java | 148 ++++++++++
.../engine/storagegroup/TsFileProcessorTest.java | 11 +
.../db/utils/datastructure/LongTVListTest.java | 8 +-
29 files changed, 1354 insertions(+), 602 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
index 299bc0b..69ebe34 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.engine.flush.pool.FlushSubTaskPoolManager;
import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.engine.memtable.IWritableMemChunk;
import org.apache.iotdb.db.exception.runtime.FlushRunTimeException;
+import org.apache.iotdb.db.nvm.datastructure.AbstractTVList;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -80,7 +81,7 @@ public class MemTableFlushTask {
long startTime = System.currentTimeMillis();
IWritableMemChunk series = memTable.getMemTableMap().get(deviceId).get(measurementId);
MeasurementSchema desc = schema.getMeasurementSchema(measurementId);
- TVList tvList = series.getSortedTVList();
+ AbstractTVList tvList = series.getSortedTVList();
sortTime += System.currentTimeMillis() - startTime;
encodingTaskQueue.add(new Pair<>(tvList, desc));
// register active time series to the ActiveTimeSeriesCounter
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
index 54b329d..08619ea 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.rescon.TVListAllocator;
import org.apache.iotdb.db.utils.MemUtils;
+import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -41,7 +42,7 @@ public abstract class AbstractMemTable implements IMemTable {
private List<Modification> modifications = new ArrayList<>();
- private final Map<String, Map<String, IWritableMemChunk>> memTableMap;
+ protected final Map<String, Map<String, IWritableMemChunk>> memTableMap;
private long memSize = 0;
@@ -63,7 +64,7 @@ public abstract class AbstractMemTable implements IMemTable {
*
* @return true if seriesPath is within this memtable
*/
- private boolean checkPath(String deviceId, String measurement) {
+ protected boolean checkPath(String deviceId, String measurement) {
return memTableMap.containsKey(deviceId) && memTableMap.get(deviceId).containsKey(measurement);
}
@@ -203,15 +204,15 @@ public abstract class AbstractMemTable implements IMemTable {
} else {
long undeletedTime = findUndeletedTime(deviceId, measurement, timeLowerBound);
IWritableMemChunk memChunk = memTableMap.get(deviceId).get(measurement);
- IWritableMemChunk chunkCopy = new WritableMemChunk(dataType, memChunk.getTVList().clone());
+ IWritableMemChunk chunkCopy = new WritableMemChunk(dataType,
+ (TVList) memChunk.getTVList().clone());
chunkCopy.setTimeOffset(undeletedTime);
sorter = chunkCopy;
}
return new ReadOnlyMemChunk(dataType, sorter, props);
}
-
- private long findUndeletedTime(String deviceId, String measurement, long timeLowerBound) {
+ protected long findUndeletedTime(String deviceId, String measurement, long timeLowerBound) {
long undeletedTime = Long.MIN_VALUE;
for (Modification modification : modifications) {
if (modification instanceof Deletion) {
@@ -242,10 +243,12 @@ public abstract class AbstractMemTable implements IMemTable {
this.modifications.add(deletion);
}
+ @Override
public void setVersion(long version) {
this.version = version;
}
+ @Override
public long getVersion() {
return version;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java
index 1739a65..0d9b183 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IWritableMemChunk.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.db.engine.memtable;
import java.util.List;
-import org.apache.iotdb.db.utils.datastructure.TVList;
+import org.apache.iotdb.db.nvm.datastructure.AbstractTVList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -68,9 +68,9 @@ public interface IWritableMemChunk extends TimeValuePairSorter {
* served for query requests.
* @return
*/
- default TVList getSortedTVList(){return null;}
+ default AbstractTVList getSortedTVList(){return null;}
- default TVList getTVList(){return null;}
+ default AbstractTVList getTVList(){return null;}
default long getMinTime() {
return Long.MIN_VALUE;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/PrimitiveMemTable.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/PrimitiveMemTable.java
index 7545d8c..d21b889 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/PrimitiveMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/PrimitiveMemTable.java
@@ -22,7 +22,9 @@ package org.apache.iotdb.db.engine.memtable;
import java.util.HashMap;
import java.util.Map;
+import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.rescon.TVListAllocator;
+import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
public class PrimitiveMemTable extends AbstractMemTable {
@@ -36,7 +38,8 @@ public class PrimitiveMemTable extends AbstractMemTable {
@Override
protected IWritableMemChunk genMemSeries(TSDataType dataType) {
- return new WritableMemChunk(dataType, TVListAllocator.getInstance().allocate(dataType));
+ return new WritableMemChunk(dataType,
+ (TVList) TVListAllocator.getInstance().allocate(dataType, false));
}
@Override
@@ -59,4 +62,20 @@ public class PrimitiveMemTable extends AbstractMemTable {
return this == obj;
}
+ @Override
+ public ReadOnlyMemChunk query(String deviceId, String measurement, TSDataType dataType,
+ Map<String, String> props, long timeLowerBound) {
+ TimeValuePairSorter sorter;
+ if (!checkPath(deviceId, measurement)) {
+ return null;
+ } else {
+ long undeletedTime = findUndeletedTime(deviceId, measurement, timeLowerBound);
+ IWritableMemChunk memChunk = memTableMap.get(deviceId).get(measurement);
+ IWritableMemChunk chunkCopy = new WritableMemChunk(dataType,
+ (TVList) memChunk.getTVList().clone());
+ chunkCopy.setTimeOffset(undeletedTime);
+ sorter = chunkCopy;
+ }
+ return new ReadOnlyMemChunk(dataType, sorter, props);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index eec8828..16a1474 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -48,7 +48,6 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.CloseTsFile
import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.qp.constant.DatetimeUtils;
import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
@@ -70,6 +69,8 @@ public class TsFileProcessor {
private static final Logger logger = LoggerFactory.getLogger(TsFileProcessor.class);
+ private boolean useNVM = false;
+
private RestorableTsFileIOWriter writer;
private Schema schema;
@@ -142,13 +143,13 @@ public class TsFileProcessor {
public boolean insert(InsertPlan insertPlan) throws QueryProcessException {
if (workMemTable == null) {
- workMemTable = MemTablePool.getInstance().getAvailableMemTable(this);
+ workMemTable = MemTablePool.getInstance().getAvailableMemTable(this, useNVM);
}
// insert insertPlan to the work memtable
workMemTable.insert(insertPlan);
- if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ if (!useNVM && IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
try {
getLogNode().write(insertPlan);
} catch (IOException e) {
@@ -172,13 +173,13 @@ public class TsFileProcessor {
Integer[] results) throws QueryProcessException {
if (workMemTable == null) {
- workMemTable = MemTablePool.getInstance().getAvailableMemTable(this);
+ workMemTable = MemTablePool.getInstance().getAvailableMemTable(this, useNVM);
}
// insert insertPlan to the work memtable
workMemTable.insertBatch(batchInsertPlan, indexes);
- if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
+ if (!useNVM && IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
try {
batchInsertPlan.setIndex(new HashSet<>(indexes));
getLogNode().write(batchInsertPlan);
@@ -404,7 +405,7 @@ public class TsFileProcessor {
writer.makeMetadataVisible();
flushingMemTables.remove(memTable);
memTable.release();
- MemTablePool.getInstance().putBack(memTable, storageGroupName);
+ MemTablePool.getInstance().putBack(memTable, storageGroupName, useNVM);
logger.debug("storage group {} flush finished, remove a memtable from flushing list, "
+ "flushing memtable list size: {}", storageGroupName, flushingMemTables.size());
} finally {
@@ -553,6 +554,14 @@ public class TsFileProcessor {
return storageGroupName;
}
+ public boolean isUseNVM() {
+ return useNVM;
+ }
+
+ public void setUseNVM(boolean useNVM) {
+ this.useNVM = useNVM;
+ }
+
/**
* get the chunk(s) in the memtable (one from work memtable and the other ones in flushing
* memtables and then compact them into one TimeValuePairSorter). Then get the related
@@ -606,5 +615,4 @@ public class TsFileProcessor {
flushQueryLock.readLock().unlock();
}
}
-
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/AbstractTVList.java
similarity index 65%
copy from server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
copy to server/src/main/java/org/apache/iotdb/db/nvm/datastructure/AbstractTVList.java
index 4e9891a..b3a6e84 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/AbstractTVList.java
@@ -1,40 +1,14 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.utils.datastructure;
-
-import static org.apache.iotdb.db.rescon.PrimitiveArrayPool.ARRAY_SIZE;
-
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.iotdb.db.rescon.PrimitiveArrayPool;
+package org.apache.iotdb.db.nvm.datastructure;
+
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
-@SuppressWarnings("unused")
-public abstract class TVList {
+public abstract class AbstractTVList {
private static final String ERR_DATATYPE_NOT_CONSISTENT = "DataType not consistent";
protected static final int SMALL_ARRAY_LENGTH = 32;
- protected List<long[]> timestamps;
protected int size;
protected long[][] sortedTimestamps;
@@ -49,24 +23,11 @@ public abstract class TVList {
protected long minTime;
- public TVList() {
- timestamps = new ArrayList<>();
- size = 0;
- minTime = Long.MIN_VALUE;
- }
-
public int size() {
return size;
}
- public long getTime(int index) {
- if (index >= size) {
- throw new ArrayIndexOutOfBoundsException(index);
- }
- int arrayIndex = index / ARRAY_SIZE;
- int elementIndex = index % ARRAY_SIZE;
- return timestamps.get(arrayIndex)[elementIndex];
- }
+ public abstract long getTime(int index);
public void putLong(long time, long value) {
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
@@ -156,44 +117,16 @@ public abstract class TVList {
protected abstract void expandValues();
- public abstract TVList clone();
+ @Override
+ public abstract AbstractTVList clone();
protected abstract void releaseLastValueArray();
- protected void releaseLastTimeArray() {
- PrimitiveArrayPool.getInstance().release(timestamps.remove(timestamps.size() - 1));
- }
+ protected abstract void releaseLastTimeArray();
- public void delete(long upperBound) {
- int newSize = 0;
- minTime = Long.MAX_VALUE;
- for (int i = 0; i < size; i++) {
- long time = getTime(i);
- if (time > upperBound) {
- set(i, newSize++);
- minTime = time < minTime ? time : minTime;
- }
- }
- size = newSize;
- // release primitive arrays that are empty
- int newArrayNum = newSize / ARRAY_SIZE;
- if (newSize % ARRAY_SIZE != 0) {
- newArrayNum ++;
- }
- for (int releaseIdx = newArrayNum; releaseIdx < timestamps.size(); releaseIdx++) {
- releaseLastTimeArray();
- releaseLastValueArray();
- }
- }
+ public abstract void delete(long upperBound);
- protected void cloneAs(TVList cloneList) {
- for (long[] timestampArray : timestamps) {
- cloneList.timestamps.add(cloneTime(timestampArray));
- }
- cloneList.size = size;
- cloneList.sorted = sorted;
- cloneList.minTime = minTime;
- }
+ protected abstract void cloneAs(AbstractTVList abstractCloneList);
public void clear() {
size = 0;
@@ -207,40 +140,17 @@ public abstract class TVList {
clearSortedValue();
}
- protected void clearTime() {
- if (timestamps != null) {
- for (long[] dataArray : timestamps) {
- PrimitiveArrayPool.getInstance().release(dataArray);
- }
- timestamps.clear();
- }
- }
+ protected abstract void clearTime();
- protected void clearSortedTime() {
- if (sortedTimestamps != null) {
- for (long[] dataArray : sortedTimestamps) {
- PrimitiveArrayPool.getInstance().release(dataArray);
- }
- sortedTimestamps = null;
- }
- }
+ protected abstract void clearSortedTime();
- abstract void clearValue();
+ protected abstract void clearValue();
- abstract void clearSortedValue();
+ protected abstract void clearSortedValue();
- protected void checkExpansion() {
- if ((size % ARRAY_SIZE) == 0) {
- expandValues();
- timestamps.add((long[]) PrimitiveArrayPool.getInstance().getPrimitiveDataListByType(TSDataType.INT64));
- }
- }
+ protected abstract void checkExpansion();
- protected long[] cloneTime(long[] array) {
- long[] cloneArray = new long[array.length];
- System.arraycopy(array, 0, cloneArray, 0, array.length);
- return cloneArray;
- }
+ protected abstract Object cloneTime(Object object);
protected void sort(int lo, int hi) {
if (sorted) {
@@ -274,31 +184,14 @@ public abstract class TVList {
}
reverseRange(lo, runHi);
} else { // Ascending
- while (runHi < hi &&getTime(runHi) >= getTime(runHi - 1))
+ while (runHi < hi &&getTime(runHi) >= getTime(runHi - 1)) {
runHi++;
+ }
}
return runHi - lo;
}
- public static TVList newList(TSDataType dataType) {
- switch (dataType) {
- case TEXT:
- return new BinaryTVList();
- case FLOAT:
- return new FloatTVList();
- case INT32:
- return new IntTVList();
- case INT64:
- return new LongTVList();
- case DOUBLE:
- return new DoubleTVList();
- case BOOLEAN:
- return new BooleanTVList();
- }
- return null;
- }
-
/**
* this field is effective only in the Tvlist in a RealOnlyMemChunk.
* @return
@@ -326,8 +219,9 @@ public abstract class TVList {
*/
protected void binarySort(int lo, int hi, int start) {
assert lo <= start && start <= hi;
- if (start == lo)
+ if (start == lo) {
start++;
+ }
for ( ; start < hi; start++) {
saveAsPivot(start);
@@ -342,10 +236,11 @@ public abstract class TVList {
*/
while (left < right) {
int mid = (left + right) >>> 1;
- if (compare(start, mid) < 0)
+ if (compare(start, mid) < 0) {
right = mid;
- else
+ } else {
left = mid + 1;
+ }
}
assert left == right;
@@ -418,7 +313,7 @@ public abstract class TVList {
}
}
- void updateMinTimeAndSorted(long[] time) {
+ protected void updateMinTimeAndSorted(long[] time) {
int length = time.length;
long inPutMinTime = Long.MAX_VALUE;
boolean inputSorted = true;
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java b/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/NVMIntTVList.java
similarity index 53%
copy from server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
copy to server/src/main/java/org/apache/iotdb/db/nvm/datastructure/NVMIntTVList.java
index 8726a9e..4ea49cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/NVMIntTVList.java
@@ -1,39 +1,24 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.utils.datastructure;
-
-import static org.apache.iotdb.db.rescon.PrimitiveArrayPool.ARRAY_SIZE;
+package org.apache.iotdb.db.nvm.datastructure;
+
+import static org.apache.iotdb.db.nvm.rescon.NVMPrimitiveArrayPool.ARRAY_SIZE;
import java.util.ArrayList;
import java.util.List;
+import org.apache.iotdb.db.nvm.rescon.NVMPrimitiveArrayPool;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager.NVMSpace;
import org.apache.iotdb.db.rescon.PrimitiveArrayPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-public class IntTVList extends TVList {
+public class NVMIntTVList extends NVMTVList {
- private List<int[]> values;
+ private List<NVMSpace> values;
+ // TODO
private int[][] sortedValues;
private int pivotValue;
- IntTVList() {
+ NVMIntTVList() {
super();
values = new ArrayList<>();
}
@@ -44,8 +29,8 @@ public class IntTVList extends TVList {
int arrayIndex = size / ARRAY_SIZE;
int elementIndex = size % ARRAY_SIZE;
minTime = minTime <= timestamp ? minTime : timestamp;
- timestamps.get(arrayIndex)[elementIndex] = timestamp;
- values.get(arrayIndex)[elementIndex] = value;
+ timestamps.get(arrayIndex).set(elementIndex, timestamp);
+ values.get(arrayIndex).set(elementIndex, value);
size++;
if (sorted && size > 1 && timestamp < getTime(size - 2)) {
sorted = false;
@@ -59,7 +44,7 @@ public class IntTVList extends TVList {
}
int arrayIndex = index / ARRAY_SIZE;
int elementIndex = index % ARRAY_SIZE;
- return values.get(arrayIndex)[elementIndex];
+ return (int) values.get(arrayIndex).get(elementIndex);
}
protected void set(int index, long timestamp, int value) {
@@ -68,26 +53,25 @@ public class IntTVList extends TVList {
}
int arrayIndex = index / ARRAY_SIZE;
int elementIndex = index % ARRAY_SIZE;
- timestamps.get(arrayIndex)[elementIndex] = timestamp;
- values.get(arrayIndex)[elementIndex] = value;
+ timestamps.get(arrayIndex).set(elementIndex, timestamp);
+ values.get(arrayIndex).set(elementIndex, value);
}
@Override
- public IntTVList clone() {
- IntTVList cloneList = new IntTVList();
+ public NVMIntTVList clone() {
+ NVMIntTVList cloneList = new NVMIntTVList();
cloneAs(cloneList);
- for (int[] valueArray : values) {
- cloneList.values.add(cloneValue(valueArray));
+ for (NVMSpace valueSpace : values) {
+ cloneList.values.add(cloneValue(valueSpace));
}
return cloneList;
}
- private int[] cloneValue(int[] array) {
- int[] cloneArray = new int[array.length];
- System.arraycopy(array, 0, cloneArray, 0, array.length);
- return cloneArray;
+ private NVMSpace cloneValue(NVMSpace valueSpace) {
+ return valueSpace.clone();
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -104,17 +88,17 @@ public class IntTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
- for (int[] dataArray : values) {
- PrimitiveArrayPool.getInstance().release(dataArray);
+ for (NVMSpace valueSpace : values) {
+ PrimitiveArrayPool.getInstance().release(valueSpace);
}
values.clear();
}
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (int[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -128,17 +112,20 @@ public class IntTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
int srcV = getInt(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getInt(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
@@ -153,7 +140,7 @@ public class IntTVList extends TVList {
@Override
protected void expandValues() {
- values.add((int[]) PrimitiveArrayPool
+ values.add(NVMPrimitiveArrayPool
.getInstance().getPrimitiveDataListByType(TSDataType.INT32));
}
@@ -179,28 +166,32 @@ public class IntTVList extends TVList {
int idx = 0;
int length = time.length;
- updateMinTimeAndSorted(time);
-
- while (idx < length) {
- int inputRemaining = length - idx;
- int arrayIdx = size / ARRAY_SIZE;
- int elementIdx = size % 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, timestamps.get(arrayIdx), elementIdx, inputRemaining);
- System.arraycopy(value, idx, values.get(arrayIdx), elementIdx, inputRemaining);
- size += 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, timestamps.get(arrayIdx), elementIdx, internalRemaining);
- System.arraycopy(value, idx, values.get(arrayIdx), elementIdx, internalRemaining);
- idx += internalRemaining;
- size += internalRemaining;
- checkExpansion();
- }
+ for (int i = 0; i < length; i++) {
+ putInt(time[i], value[i]);
}
+
+// updateMinTimeAndSorted(time);
+//
+// while (idx < length) {
+// int inputRemaining = length - idx;
+// int arrayIdx = size / ARRAY_SIZE;
+// int elementIdx = size % 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, timestamps.get(arrayIdx), elementIdx, inputRemaining);
+// System.arraycopy(value, idx, values.get(arrayIdx), elementIdx, inputRemaining);
+// size += 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, timestamps.get(arrayIdx), elementIdx, internalRemaining);
+// System.arraycopy(value, idx, values.get(arrayIdx), elementIdx, internalRemaining);
+// idx += internalRemaining;
+// size += internalRemaining;
+// checkExpansion();
+// }
+// }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/NVMTVList.java b/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/NVMTVList.java
new file mode 100644
index 0000000..2b83735
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/datastructure/NVMTVList.java
@@ -0,0 +1,123 @@
+package org.apache.iotdb.db.nvm.datastructure;
+
+import static org.apache.iotdb.db.nvm.rescon.NVMPrimitiveArrayPool.ARRAY_SIZE;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.db.nvm.rescon.NVMPrimitiveArrayPool;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager.NVMSpace;
+import org.apache.iotdb.db.rescon.PrimitiveArrayPool;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+public abstract class NVMTVList extends AbstractTVList {
+
+ protected List<NVMSpace> timestamps;
+
+ public NVMTVList() {
+ timestamps = new ArrayList<>();
+ size = 0;
+ minTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public long getTime(int index) {
+ if (index >= size) {
+ throw new ArrayIndexOutOfBoundsException(index);
+ }
+ int arrayIndex = index / ARRAY_SIZE;
+ int elementIndex = index % ARRAY_SIZE;
+ return (long) timestamps.get(arrayIndex).get(elementIndex);
+ }
+
+ @Override
+ protected void releaseLastTimeArray() {
+ NVMPrimitiveArrayPool.getInstance().release(timestamps.remove(timestamps.size() - 1), TSDataType.INT64);
+ }
+
+ @Override
+ public void delete(long upperBound) {
+ int newSize = 0;
+ minTime = Long.MAX_VALUE;
+ for (int i = 0; i < size; i++) {
+ long time = getTime(i);
+ if (time > upperBound) {
+ set(i, newSize++);
+ minTime = time < minTime ? time : minTime;
+ }
+ }
+ size = newSize;
+ // release primitive arrays that are empty
+ int newArrayNum = newSize / ARRAY_SIZE;
+ if (newSize % ARRAY_SIZE != 0) {
+ newArrayNum ++;
+ }
+ for (int releaseIdx = newArrayNum; releaseIdx < timestamps.size(); releaseIdx++) {
+ releaseLastTimeArray();
+ releaseLastValueArray();
+ }
+ }
+
+ @Override
+ protected void cloneAs(AbstractTVList abstractCloneList) {
+ NVMTVList cloneList = (NVMTVList) abstractCloneList;
+ for (NVMSpace timeSpace : timestamps) {
+ cloneList.timestamps.add((NVMSpace) cloneTime(timeSpace));
+ }
+ cloneList.size = size;
+ cloneList.sorted = sorted;
+ cloneList.minTime = minTime;
+ }
+
+ @Override
+ protected void clearTime() {
+ if (timestamps != null) {
+ for (NVMSpace timeSpace : timestamps) {
+ NVMPrimitiveArrayPool.getInstance().release(timeSpace, TSDataType.INT64);
+ }
+ timestamps.clear();
+ }
+ }
+
+ @Override
+ protected void clearSortedTime() {
+ if (sortedTimestamps != null) {
+ for (long[] dataArray : sortedTimestamps) {
+ PrimitiveArrayPool.getInstance().release(dataArray);
+ }
+ sortedTimestamps = null;
+ }
+ }
+
+ @Override
+ protected void checkExpansion() {
+ if ((size % ARRAY_SIZE) == 0) {
+ expandValues();
+ timestamps.add(NVMPrimitiveArrayPool.getInstance().getPrimitiveDataListByType(TSDataType.INT64));
+ }
+ }
+
+ @Override
+ protected Object cloneTime(Object object) {
+ NVMSpace timeSpace = (NVMSpace) object;
+ return timeSpace.clone();
+ }
+
+ public static NVMTVList newList(TSDataType dataType) {
+ switch (dataType) {
+ case TEXT:
+ // TODO
+// return new BinaryTVList();
+ case FLOAT:
+// return new FloatTVList();
+ case INT32:
+ return new NVMIntTVList();
+ case INT64:
+// return new LongTVList();
+ case DOUBLE:
+// return new DoubleTVList();
+ case BOOLEAN:
+// return new BooleanTVList();
+ }
+ return null;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/executor/INVMExecutor.java b/server/src/main/java/org/apache/iotdb/db/nvm/executor/INVMExecutor.java
new file mode 100644
index 0000000..52c8541
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/executor/INVMExecutor.java
@@ -0,0 +1,12 @@
+//package org.apache.iotdb.db.nvm.executor;
+//
+//public interface INVMExecutor {
+//
+// void flushRecordToNVM(Record record);
+//
+// void commit(Transaction transaction);
+//
+// void abort(Transaction transaction);
+//
+// void recovery();
+//}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/executor/NVMExecutor.java b/server/src/main/java/org/apache/iotdb/db/nvm/executor/NVMExecutor.java
new file mode 100644
index 0000000..3fa5c2d
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/executor/NVMExecutor.java
@@ -0,0 +1,134 @@
+//package org.apache.iotdb.db.nvm.executor;
+//
+//import java.io.IOException;
+//import java.io.RandomAccessFile;
+//import java.nio.channels.FileChannel;
+//import java.nio.channels.FileChannel.MapMode;
+//import java.util.HashMap;
+//import java.util.HashSet;
+//import java.util.List;
+//import java.util.Map;
+//import java.util.Set;
+//import org.apache.iotdb.tsfile.utils.Pair;
+//import org.slf4j.Logger;
+//import org.slf4j.LoggerFactory;
+//
+//public class NVMExecutor implements INVMExecutor {
+//
+// private static final Logger logger = LoggerFactory.getLogger(NVMExecutor.class);
+//
+// // 逻辑PageID -> PCM中对应脏页及每个脏页对应的脏record
+// private Map<Long, Pair<Long, Set<Long>>> mappingTable;
+//
+// private final long FREE_SLOT_BITMAP_BUF_SIZE = 1000;
+// private NVMSpace freeSlotBitmapBuf;
+//
+// private final long ACTIVE_TX_LIST_BUF_SIZE = 1000;
+// private NVMSpace activeTxListBuf;
+//
+// private final long DATA_BUF_SIZE = 1000;
+// private NVMSpace dataBuf;
+//
+// // 未提交的事务XID -> 所有脏record
+// private Map<Long, Set<Long>> transactionTable;
+// // 被未提交事务修改的page PID -> 已提交版本page及对应record的已提交版本
+// private Map<Long, Pair<Long, Set<Long>>> dirtyPageTable;
+// // 每个page PID -> DRAM中的脏record列表
+// private Map<Long, List<Long>> dirtyRecordTable;
+//
+// private NVMExecutor() {
+// initMappingTable();
+//
+// initNVMFields();
+//
+// transactionTable = new HashMap<>();
+// dirtyPageTable = new HashMap<>();
+// dirtyRecordTable = new HashMap<>();
+// }
+//
+// private static final NVMExecutor INSTANCE = new NVMExecutor();
+// public static NVMExecutor getInstance() {
+// return INSTANCE;
+// }
+//
+// /**
+// * init by inverse table
+// */
+// private void initMappingTable() {
+// // TODO
+// }
+//
+// private void initNVMFields() {
+// FileChannel fc = null;
+// try {
+// fc = new RandomAccessFile(NVM_PATH, "rw").getChannel();
+//
+// freeSlotBitmapBuf = fc.map(MapMode.READ_WRITE, FREE_SLOT_BITMAP_BUF_OFFSET, FREE_SLOT_BITMAP_BUF_OFFSET);
+// activeTxListBuf = fc.map(MapMode.READ_WRITE, ACTIVE_TX_LIST_BUF_OFFSET, ACTIVE_TX_LIST_BUF_SIZE);
+// dataBuf = fc.map(MapMode.READ_WRITE, DATA_BUF_OFFSET, DATA_BUF_SIZE);
+// } catch (IOException e) {
+// logger.error("Fail to init NVM fields at {}.", NVM_PATH, e);
+// }
+// }
+//
+// @Override
+// public void flushRecordToNVM(Record record) {
+// /*
+// Let T be the transaction that made the last update to t
+// if t is T’s first dirty record to be flushed then
+// Append T to the ActiveTxList in PCM
+// Write t to a free space of PCM
+// if there is a previously committed version of t in PCM then
+// Add the previous version to the Dirty Page Table
+// else if there is a copy of t (updated by T) in PCM then
+// Invalidate the uncommitted copy
+// Update the Mapping Table and Transaction Table */
+//
+// if (isFirstToNVM(record.getXID())) {
+// addActiveTx(record.getXID());
+// }
+//
+// writeRecord(record);
+//
+// Record committedRecord = getPreviousComittedRecord(record);
+// if (committedRecord != null) {
+// if (!dirtyPageTable.containsKey(record.getPID())) {
+// dirtyPageTable.put(record.getPID(), new Pair<>(committedRecord.getPID(), new HashSet<>()));
+// }
+// dirtyPageTable.get(record.getPID()).right.add(record.getRID());
+// }
+// }
+//
+// private Record getPreviousComittedRecord(Record record) {
+// // TODO
+// return null;
+// }
+//
+// private void writeRecord(Record record) {
+// // TODO
+// }
+//
+// private void addActiveTx(long XID) {
+// // TODO
+// }
+//
+// private boolean isFirstToNVM(long XID) {
+// // TODO
+// return true;
+// }
+//
+// @Override
+// public void commit(Transaction transaction) {
+//
+// }
+//
+// @Override
+// public void abort(Transaction transaction) {
+//
+// }
+//
+// @Override
+// public void recovery() {
+//
+// }
+//}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/memtable/NVMPrimitiveMemTable.java b/server/src/main/java/org/apache/iotdb/db/nvm/memtable/NVMPrimitiveMemTable.java
new file mode 100644
index 0000000..bce7fad
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/memtable/NVMPrimitiveMemTable.java
@@ -0,0 +1,65 @@
+package org.apache.iotdb.db.nvm.memtable;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.iotdb.db.engine.memtable.AbstractMemTable;
+import org.apache.iotdb.db.engine.memtable.IMemTable;
+import org.apache.iotdb.db.engine.memtable.IWritableMemChunk;
+import org.apache.iotdb.db.engine.memtable.TimeValuePairSorter;
+import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.nvm.datastructure.NVMTVList;
+import org.apache.iotdb.db.rescon.TVListAllocator;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+public class NVMPrimitiveMemTable extends AbstractMemTable {
+
+ public NVMPrimitiveMemTable() {
+ }
+
+ public NVMPrimitiveMemTable(Map<String, Map<String, IWritableMemChunk>> memTableMap) {
+ super(memTableMap);
+ }
+
+ @Override
+ protected IWritableMemChunk genMemSeries(TSDataType dataType) {
+ return new NVMWritableMemChunk(dataType,
+ (NVMTVList) TVListAllocator.getInstance().allocate(dataType, true));
+ }
+
+ @Override
+ public IMemTable copy() {
+ Map<String, Map<String, IWritableMemChunk>> newMap = new HashMap<>(getMemTableMap());
+
+ return new NVMPrimitiveMemTable(newMap);
+ }
+
+ @Override
+ public boolean isSignalMemTable() {
+ return false;
+ }
+
+ @Override
+ public int hashCode() {return (int) getVersion();}
+
+ @Override
+ public boolean equals(Object obj) {
+ return this == obj;
+ }
+
+ @Override
+ public ReadOnlyMemChunk query(String deviceId, String measurement, TSDataType dataType,
+ Map<String, String> props, long timeLowerBound) {
+ TimeValuePairSorter sorter;
+ if (!checkPath(deviceId, measurement)) {
+ return null;
+ } else {
+ long undeletedTime = findUndeletedTime(deviceId, measurement, timeLowerBound);
+ IWritableMemChunk memChunk = memTableMap.get(deviceId).get(measurement);
+ IWritableMemChunk chunkCopy = new NVMWritableMemChunk(dataType,
+ (NVMTVList) memChunk.getTVList().clone());
+ chunkCopy.setTimeOffset(undeletedTime);
+ sorter = chunkCopy;
+ }
+ return new ReadOnlyMemChunk(dataType, sorter, props);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/memtable/NVMWritableMemChunk.java b/server/src/main/java/org/apache/iotdb/db/nvm/memtable/NVMWritableMemChunk.java
new file mode 100644
index 0000000..4c7f287
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/memtable/NVMWritableMemChunk.java
@@ -0,0 +1,269 @@
+package org.apache.iotdb.db.nvm.memtable;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.iotdb.db.engine.memtable.IWritableMemChunk;
+import org.apache.iotdb.db.nvm.datastructure.NVMTVList;
+import org.apache.iotdb.db.utils.TimeValuePair;
+import org.apache.iotdb.db.utils.TsPrimitiveType.TsBinary;
+import org.apache.iotdb.db.utils.TsPrimitiveType.TsBoolean;
+import org.apache.iotdb.db.utils.TsPrimitiveType.TsDouble;
+import org.apache.iotdb.db.utils.TsPrimitiveType.TsFloat;
+import org.apache.iotdb.db.utils.TsPrimitiveType.TsInt;
+import org.apache.iotdb.db.utils.TsPrimitiveType.TsLong;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NVMWritableMemChunk implements IWritableMemChunk {
+
+ private static final Logger logger = LoggerFactory.getLogger(NVMWritableMemChunk.class);
+ private TSDataType dataType;
+ private NVMTVList list;
+ private List<TimeValuePair> sortedList;
+
+ public NVMWritableMemChunk(TSDataType dataType, NVMTVList list) {
+ this.dataType = dataType;
+ this.list = list;
+ }
+
+ @Override
+ public void write(long insertTime, Object objectValue) {
+ switch (dataType) {
+ case BOOLEAN:
+ putBoolean(insertTime, (boolean) objectValue);
+ break;
+ case INT32:
+ putInt(insertTime, (int) objectValue);
+ break;
+ case INT64:
+ putLong(insertTime, (long) objectValue);
+ break;
+ case FLOAT:
+ putFloat(insertTime, (float) objectValue);
+ break;
+ case DOUBLE:
+ putDouble(insertTime, (double) objectValue);
+ break;
+ case TEXT:
+ putBinary(insertTime, (Binary) objectValue);
+ break;
+ default:
+ throw new UnSupportedDataTypeException("Unsupported data type:" + dataType);
+ }
+ sortedList = null;
+ }
+
+ @Override
+ public void write(long[] times, Object valueList, TSDataType dataType, List<Integer> indexes) {
+ switch (dataType) {
+ case BOOLEAN:
+ boolean[] boolValues = (boolean[]) valueList;
+ if (times.length == indexes.size()) {
+ putBooleans(times, boolValues);
+ break;
+ }
+ for (Integer index : indexes) {
+ putBoolean(times[index], boolValues[index]);
+ }
+ break;
+ case INT32:
+ int[] intValues = (int[]) valueList;
+ if (times.length == indexes.size()) {
+ putInts(times, intValues);
+ break;
+ }
+ for (Integer index : indexes) {
+ putInt(times[index], intValues[index]);
+ }
+ break;
+ case INT64:
+ long[] longValues = (long[]) valueList;
+ if (times.length == indexes.size()) {
+ putLongs(times, longValues);
+ break;
+ }
+ for (Integer index : indexes) {
+ putLong(times[index], longValues[index]);
+ }
+ break;
+ case FLOAT:
+ float[] floatValues = (float[]) valueList;
+ if (times.length == indexes.size()) {
+ putFloats(times, floatValues);
+ break;
+ }
+ for (Integer index : indexes) {
+ putFloat(times[index], floatValues[index]);
+ }
+ break;
+ case DOUBLE:
+ double[] doubleValues = (double[]) valueList;
+ if (times.length == indexes.size()) {
+ putDoubles(times, doubleValues);
+ break;
+ }
+ for (Integer index : indexes) {
+ putDouble(times[index], doubleValues[index]);
+ }
+ break;
+ case TEXT:
+ Binary[] binaryValues = (Binary[]) valueList;
+ if (times.length == indexes.size()) {
+ putBinaries(times, binaryValues);
+ break;
+ }
+ for (Integer index : indexes) {
+ putBinary(times[index], binaryValues[index]);
+ }
+ break;
+ default:
+ throw new UnSupportedDataTypeException("Unsupported data type:" + dataType);
+ }
+ sortedList = null;
+ }
+
+
+ @Override
+ public void putLong(long t, long v) {
+ list.putLong(t, v);
+ }
+
+ @Override
+ public void putInt(long t, int v) {
+ list.putInt(t, v);
+ }
+
+ @Override
+ public void putFloat(long t, float v) {
+ list.putFloat(t, v);
+ }
+
+ @Override
+ public void putDouble(long t, double v) {
+ list.putDouble(t, v);
+ }
+
+ @Override
+ public void putBinary(long t, Binary v) {
+ list.putBinary(t, v);
+ }
+
+ @Override
+ public void putBoolean(long t, boolean v) {
+ list.putBoolean(t, v);
+ }
+
+ @Override
+ public void putLongs(long[] t, long[] v) {
+ list.putLongs(t, v);
+ }
+
+ @Override
+ public void putInts(long[] t, int[] v) {
+ list.putInts(t, v);
+ }
+
+ @Override
+ public void putFloats(long[] t, float[] v) {
+ list.putFloats(t, v);
+ }
+
+ @Override
+ public void putDoubles(long[] t, double[] v) {
+ list.putDoubles(t, v);
+ }
+
+ @Override
+ public void putBinaries(long[] t, Binary[] v) {
+ list.putBinaries(t, v);
+ }
+
+ @Override
+ public void putBooleans(long[] t, boolean[] v) {
+ list.putBooleans(t, v);
+ }
+
+ @Override
+ public synchronized NVMTVList getSortedTVList() {
+ list.sort();
+ return list;
+ }
+
+ @Override
+ public NVMTVList getTVList() {
+ return list;
+ }
+
+ @Override
+ public long count() {
+ return list.size();
+ }
+
+ @Override
+ public TSDataType getType() {
+ return dataType;
+ }
+
+ @Override
+ public void setTimeOffset(long offset) {
+ list.setTimeOffset(offset);
+ }
+
+ @Override
+ public synchronized List<TimeValuePair> getSortedTimeValuePairList() {
+ if (sortedList != null) {
+ return sortedList;
+ }
+ sortedList = new ArrayList<>();
+ list.sort();
+ for (int i = 0; i < list.size(); i++) {
+ long time = list.getTime(i);
+ if (time < list.getTimeOffset() ||
+ (i + 1 < list.size() && (time == list.getTime(i + 1)))) {
+ continue;
+ }
+ switch (dataType) {
+ case BOOLEAN:
+ sortedList.add(new TimeValuePair(time, new TsBoolean(list.getBoolean(i))));
+ break;
+ case INT32:
+ sortedList.add(new TimeValuePair(time, new TsInt(list.getInt(i))));
+ break;
+ case INT64:
+ sortedList.add(new TimeValuePair(time, new TsLong(list.getLong(i))));
+ break;
+ case FLOAT:
+ sortedList.add(new TimeValuePair(time, new TsFloat(list.getFloat(i))));
+ break;
+ case DOUBLE:
+ sortedList.add(new TimeValuePair(time, new TsDouble(list.getDouble(i))));
+ break;
+ case TEXT:
+ sortedList.add(new TimeValuePair(time, new TsBinary(list.getBinary(i))));
+ break;
+ default:
+ logger.error("Unsupported data type: {}", dataType);
+ break;
+ }
+ }
+ return this.sortedList;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return list.size() == 0;
+ }
+
+ @Override
+ public long getMinTime() {
+ return list.getMinTime();
+ }
+
+ @Override
+ public void delete(long upperBound) {
+ list.delete(upperBound);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/rescon/NVMPrimitiveArrayPool.java b/server/src/main/java/org/apache/iotdb/db/nvm/rescon/NVMPrimitiveArrayPool.java
new file mode 100644
index 0000000..0647217
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/rescon/NVMPrimitiveArrayPool.java
@@ -0,0 +1,68 @@
+package org.apache.iotdb.db.nvm.rescon;
+
+import java.util.ArrayDeque;
+import java.util.EnumMap;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager.NVMSpace;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+public class NVMPrimitiveArrayPool {
+
+ /**
+ * data type -> Array<PrimitiveArray>
+ */
+ private static final EnumMap<TSDataType, ArrayDeque<NVMSpace>> primitiveArraysMap = new EnumMap<>(TSDataType.class);
+
+ public static final int ARRAY_SIZE = 128;
+
+ static {
+ primitiveArraysMap.put(TSDataType.BOOLEAN, new ArrayDeque());
+ primitiveArraysMap.put(TSDataType.INT32, new ArrayDeque());
+ primitiveArraysMap.put(TSDataType.INT64, new ArrayDeque());
+ primitiveArraysMap.put(TSDataType.FLOAT, new ArrayDeque());
+ primitiveArraysMap.put(TSDataType.DOUBLE, new ArrayDeque());
+ primitiveArraysMap.put(TSDataType.TEXT, new ArrayDeque());
+ }
+
+ public static NVMPrimitiveArrayPool getInstance() {
+ return INSTANCE;
+ }
+
+ private static final NVMPrimitiveArrayPool INSTANCE = new NVMPrimitiveArrayPool();
+
+
+ private NVMPrimitiveArrayPool() {}
+
+ public synchronized NVMSpace getPrimitiveDataListByType(TSDataType dataType) {
+ ArrayDeque<NVMSpace> dataListQueue = primitiveArraysMap.computeIfAbsent(dataType, k ->new ArrayDeque<>());
+ NVMSpace nvmSpace = dataListQueue.poll();
+
+ long size = NVMSpaceManager.getPrimitiveTypeByteSize(dataType);
+ if (nvmSpace == null) {
+ nvmSpace = NVMSpaceManager.allocate(size * ARRAY_SIZE, dataType);
+ }
+
+ return nvmSpace;
+ }
+
+
+ public synchronized void release(NVMSpace nvmSpace, TSDataType dataType) {
+ // TODO freeslotmap?
+
+ primitiveArraysMap.get(dataType).add(nvmSpace);
+ }
+
+ /**
+ * @param size needed capacity
+ * @return an array of primitive data arrays
+ */
+ public synchronized NVMSpace[] getDataListsByType(TSDataType dataType, int size) {
+ int arrayNumber = (int) Math.ceil((float) size / (float) ARRAY_SIZE);
+ NVMSpace[] nvmSpaces = new NVMSpace[arrayNumber];
+ for (int i = 0; i < arrayNumber; i++) {
+ nvmSpaces[i] = getPrimitiveDataListByType(dataType);
+ }
+ return nvmSpaces;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/space/INVMSpaceManager.java b/server/src/main/java/org/apache/iotdb/db/nvm/space/INVMSpaceManager.java
new file mode 100644
index 0000000..de56fa4
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/space/INVMSpaceManager.java
@@ -0,0 +1,9 @@
+package org.apache.iotdb.db.nvm.space;
+
+import java.io.IOException;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager.NVMSpace;
+
+public interface INVMSpaceManager {
+
+ NVMSpace allocate(long size) throws IOException;
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/nvm/space/NVMSpaceManager.java b/server/src/main/java/org/apache/iotdb/db/nvm/space/NVMSpaceManager.java
new file mode 100644
index 0000000..d5e70a1
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/nvm/space/NVMSpaceManager.java
@@ -0,0 +1,174 @@
+package org.apache.iotdb.db.nvm.space;
+
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileChannel.MapMode;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NVMSpaceManager {
+
+ private static final Logger logger = LoggerFactory.getLogger(NVMSpaceManager.class);
+
+ private static String NVM_PATH;
+ private static FileChannel nvmFileChannel;
+ private static final MapMode MAP_MODE = MapMode.READ_WRITE;
+ private static long nvmSize;
+ private static AtomicLong curOffset = new AtomicLong(0L);
+
+ public static void init(String path) {
+ try {
+ NVM_PATH = path;
+ nvmFileChannel = new RandomAccessFile(NVM_PATH, "rw").getChannel();
+ nvmSize = nvmFileChannel.size();
+ } catch (IOException e) {
+ logger.error("Fail to open NVM space at {}.", NVM_PATH, e);
+ // TODO deal with error
+ }
+ }
+
+ public static int getPrimitiveTypeByteSize(TSDataType dataType) {
+ int size = 0;
+ switch (dataType) {
+ case BOOLEAN:
+ size = 8;
+ break;
+ case INT32:
+ size = Integer.SIZE;
+ break;
+ case INT64:
+ size = Long.SIZE;
+ break;
+ case FLOAT:
+ size = Float.SIZE;
+ break;
+ case DOUBLE:
+ size = Double.SIZE;
+ break;
+ case TEXT:
+ // TODO
+ break;
+ default:
+ throw new UnSupportedDataTypeException("DataType: " + dataType);
+ }
+ return size >> 3;
+ }
+
+ public static NVMSpace allocate(long size, TSDataType dataType) {
+ long offset = curOffset.getAndAdd(size);
+ if (offset + size > nvmSize) {
+ // TODO throw exception
+ }
+
+ try {
+ return new NVMSpace(offset, size, nvmFileChannel.map(MAP_MODE, offset, size), dataType);
+ } catch (IOException e) {
+ // TODO deal with error
+ e.printStackTrace();
+ return null;
+ }
+ }
+
+ public static class NVMSpace {
+
+ private long offset;
+ private long size;
+ private ByteBuffer byteBuffer;
+ private TSDataType dataType;
+
+ private NVMSpace(long offset, long size, ByteBuffer byteBuffer, TSDataType dataType) {
+ this.offset = offset;
+ this.size = size;
+ this.byteBuffer = byteBuffer;
+ this.dataType = dataType;
+ }
+
+ public long getOffset() {
+ return offset;
+ }
+
+ public long getSize() {
+ return size;
+ }
+
+ public ByteBuffer getByteBuffer() {
+ return byteBuffer;
+ }
+
+ @Override
+ public NVMSpace clone() {
+ NVMSpace cloneSpace = NVMSpaceManager.allocate(size, dataType);
+ int position = byteBuffer.position();
+ byteBuffer.rewind();
+ cloneSpace.getByteBuffer().put(byteBuffer);
+ byteBuffer.position(position);
+ cloneSpace.getByteBuffer().flip();
+ return cloneSpace;
+ }
+
+ public Object get(int index) {
+ int objectSize = NVMSpaceManager.getPrimitiveTypeByteSize(dataType);
+ index *= objectSize;
+ Object object = null;
+ switch (dataType) {
+ case BOOLEAN:
+ object = byteBuffer.get(index);
+ break;
+ case INT32:
+ object = byteBuffer.getInt(index);
+ break;
+ case INT64:
+ object = byteBuffer.getLong(index);
+ break;
+ case FLOAT:
+ object = byteBuffer.getFloat(index);
+ break;
+ case DOUBLE:
+ object = byteBuffer.getDouble(index);
+ break;
+ case TEXT:
+ // TODO
+ break;
+ }
+ return object;
+ }
+
+ public void set(int index, Object object) {
+ int objectSize = NVMSpaceManager.getPrimitiveTypeByteSize(dataType);
+ index *= objectSize;
+ switch (dataType) {
+ case BOOLEAN:
+ object = byteBuffer.put(index, (byte) object);
+ break;
+ case INT32:
+ object = byteBuffer.putInt(index, (int) object);
+ break;
+ case INT64:
+ object = byteBuffer.putLong(index, (long) object);
+ break;
+ case FLOAT:
+ object = byteBuffer.putFloat(index, (float) object);
+ break;
+ case DOUBLE:
+ object = byteBuffer.putDouble(index, (double) object);
+ break;
+ case TEXT:
+ // TODO
+ break;
+ }
+ }
+ }
+
+ public static String getNvmPath() {
+ return NVM_PATH;
+ }
+
+ public static void setNvmPath(String nvmPath) {
+ NVM_PATH = nvmPath;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java b/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java
index 4b03cb1..83211ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/MemTablePool.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
+import org.apache.iotdb.db.nvm.memtable.NVMPrimitiveMemTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,8 +35,10 @@ public class MemTablePool {
private static final Logger logger = LoggerFactory.getLogger(MemTablePool.class);
private static final Deque<IMemTable> availableMemTables = new ArrayDeque<>();
+ private static final Deque<IMemTable> availableNVMMemTables = new ArrayDeque<>();
private int size = 0;
+ private int nvmSize = 0;
private static final int WAIT_TIME = 2000;
@@ -43,62 +46,118 @@ public class MemTablePool {
}
// TODO change the impl of getAvailableMemTable to non-blocking
- public IMemTable getAvailableMemTable(Object applier) {
- synchronized (availableMemTables) {
- if (availableMemTables.isEmpty() && size < CONFIG.getMaxMemtableNumber()) {
- size++;
- logger.info("generated a new memtable for {}, system memtable size: {}, stack size: {}",
- applier, size, availableMemTables.size());
- return new PrimitiveMemTable();
- } else if (!availableMemTables.isEmpty()) {
- logger
- .debug("system memtable size: {}, stack size: {}, then get a memtable from stack for {}",
- size, availableMemTables.size(), applier);
- return availableMemTables.pop();
- }
+ public IMemTable getAvailableMemTable(Object applier, boolean nvm) {
+ if (nvm) {
+ synchronized (availableNVMMemTables) {
+ if (availableNVMMemTables.isEmpty() && nvmSize < CONFIG.getMaxMemtableNumber()) {
+ nvmSize++;
+ logger.info("generated a new nvm memtable for {}, system memtable size: {}, stack size: {}",
+ applier, nvmSize, availableNVMMemTables.size());
+ return new NVMPrimitiveMemTable();
+ } else if (!availableNVMMemTables.isEmpty()) {
+ logger
+ .debug(
+ "system nvm memtable size: {}, stack size: {}, then get a nvm memtable from stack for {}",
+ nvmSize, availableNVMMemTables.size(), applier);
+ return availableNVMMemTables.pop();
+ }
- // wait until some one has released a memtable
- int waitCount = 1;
- while (true) {
- if (!availableMemTables.isEmpty()) {
- logger.debug(
- "system memtable size: {}, stack size: {}, then get a memtable from stack for {}",
- size, availableMemTables.size(), applier);
+ // wait until some one has released a memtable
+ int waitCount = 1;
+ while (true) {
+ if (!availableNVMMemTables.isEmpty()) {
+ logger.debug(
+ "system nvm memtable size: {}, stack size: {}, then get a memtable from stack for {}",
+ nvmSize, availableNVMMemTables.size(), applier);
+ return availableNVMMemTables.pop();
+ }
+ try {
+ availableNVMMemTables.wait(WAIT_TIME);
+ } catch (InterruptedException e) {
+ logger.error("{} fails to wait fot nvm memtables {}, continue to wait", applier, e);
+ Thread.currentThread().interrupt();
+ }
+ logger.info("{} has waited for a nvm memtable for {}ms", applier, waitCount++ * WAIT_TIME);
+ }
+ }
+ } else {
+ synchronized (availableMemTables) {
+ if (availableMemTables.isEmpty() && size < CONFIG.getMaxMemtableNumber()) {
+ size++;
+ logger.info("generated a new memtable for {}, system memtable size: {}, stack size: {}",
+ applier, size, availableMemTables.size());
+ return new PrimitiveMemTable();
+ } else if (!availableMemTables.isEmpty()) {
+ logger
+ .debug(
+ "system memtable size: {}, stack size: {}, then get a memtable from stack for {}",
+ size, availableMemTables.size(), applier);
return availableMemTables.pop();
}
- try {
- availableMemTables.wait(WAIT_TIME);
- } catch (InterruptedException e) {
- logger.error("{} fails to wait fot memtables {}, continue to wait", applier, e);
- Thread.currentThread().interrupt();
+
+ // wait until some one has released a memtable
+ int waitCount = 1;
+ while (true) {
+ if (!availableMemTables.isEmpty()) {
+ logger.debug(
+ "system memtable size: {}, stack size: {}, then get a memtable from stack for {}",
+ size, availableMemTables.size(), applier);
+ return availableMemTables.pop();
+ }
+ try {
+ availableMemTables.wait(WAIT_TIME);
+ } catch (InterruptedException e) {
+ logger.error("{} fails to wait fot memtables {}, continue to wait", applier, e);
+ Thread.currentThread().interrupt();
+ }
+ logger.info("{} has waited for a memtable for {}ms", applier, waitCount++ * WAIT_TIME);
}
- logger.info("{} has waited for a memtable for {}ms", applier, waitCount++ * WAIT_TIME);
}
}
}
- public void putBack(IMemTable memTable, String storageGroup) {
+ public void putBack(IMemTable memTable, String storageGroup, boolean nvm) {
if (memTable.isSignalMemTable()) {
return;
}
- synchronized (availableMemTables) {
- // because of dynamic parameter adjust, the max number of memtable may decrease.
- if (size > CONFIG.getMaxMemtableNumber()) {
- logger.debug(
- "Currently the size of available MemTables is {}, the maxmin size of MemTables is {}, discard this MemTable.",
- CONFIG.getMaxMemtableNumber(), size);
- size--;
- return;
+
+ if (nvm) {
+ synchronized (availableNVMMemTables) {
+ // because of dynamic parameter adjust, the max number of memtable may decrease.
+ if (nvmSize > CONFIG.getMaxMemtableNumber()) {
+ logger.debug(
+ "Currently the size of available nvm MemTables is {}, the maxmin size of nvm MemTables is {}, discard this nvm MemTable.",
+ CONFIG.getMaxMemtableNumber(), nvmSize);
+ nvmSize--;
+ return;
+ }
+ memTable.clear();
+ availableNVMMemTables.push(memTable);
+ availableNVMMemTables.notify();
+ logger
+ .debug("{} return a nvm memtable, stack size {}", storageGroup, availableNVMMemTables.size());
+ }
+ } else {
+ synchronized (availableMemTables) {
+ // because of dynamic parameter adjust, the max number of memtable may decrease.
+ if (size > CONFIG.getMaxMemtableNumber()) {
+ logger.debug(
+ "Currently the size of available MemTables is {}, the maxmin size of MemTables is {}, discard this MemTable.",
+ CONFIG.getMaxMemtableNumber(), size);
+ size--;
+ return;
+ }
+ memTable.clear();
+ availableMemTables.push(memTable);
+ availableMemTables.notify();
+ logger
+ .debug("{} return a memtable, stack size {}", storageGroup, availableMemTables.size());
}
- memTable.clear();
- availableMemTables.push(memTable);
- availableMemTables.notify();
- logger.debug("{} return a memtable, stack size {}", storageGroup, availableMemTables.size());
}
}
- public int getSize() {
- return size;
+ public int getSize(boolean nvm) {
+ return nvm ? nvmSize : size;
}
public static MemTablePool getInstance() {
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocator.java b/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocator.java
index b09c05a..6b90a93 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocator.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocator.java
@@ -25,6 +25,9 @@ import java.util.Map;
import java.util.Queue;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.nvm.datastructure.AbstractTVList;
+import org.apache.iotdb.db.nvm.datastructure.NVMIntTVList;
+import org.apache.iotdb.db.nvm.datastructure.NVMTVList;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.JMXService;
import org.apache.iotdb.db.service.ServiceType;
@@ -39,51 +42,76 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
public class TVListAllocator implements TVListAllocatorMBean, IService {
- private Map<TSDataType, Queue<TVList>> tvListCache = new EnumMap<>(TSDataType.class);
- private String mbeanName = String
+ protected Map<TSDataType, Queue<TVList>> tvListCache = new EnumMap<>(TSDataType.class);
+ protected Map<TSDataType, Queue<NVMTVList>> nvmTVListCache = new EnumMap<>(TSDataType.class);
+ protected String mbeanName = String
.format("%s:%s=%s", IoTDBConstant.IOTDB_PACKAGE, IoTDBConstant.JMX_TYPE,
getID().getJmxName());
- private static final TVListAllocator INSTANCE = new TVListAllocator();
+ protected static final TVListAllocator INSTANCE = new TVListAllocator();
public static TVListAllocator getInstance() {
return INSTANCE;
}
- public synchronized TVList allocate(TSDataType dataType) {
- Queue<TVList> tvLists = tvListCache.computeIfAbsent(dataType,
- k -> new ArrayDeque<>());
- TVList list = tvLists.poll();
- return list != null ? list : TVList.newList(dataType);
+ public synchronized AbstractTVList allocate(TSDataType dataType, boolean nvm) {
+ AbstractTVList list = null;
+ if (nvm) {
+ Queue<NVMTVList> tvLists = nvmTVListCache.computeIfAbsent(dataType,
+ k -> new ArrayDeque<>());
+ list = tvLists.poll();
+ return list != null ? list : NVMTVList.newList(dataType);
+ } else {
+ Queue<TVList> tvLists = tvListCache.computeIfAbsent(dataType,
+ k -> new ArrayDeque<>());
+ list = tvLists.poll();
+ return list != null ? list : TVList.newList(dataType);
+ }
}
- public synchronized void release(TSDataType dataType, TVList list) {
+ public synchronized void release(TSDataType dataType, AbstractTVList list, boolean nvm) {
list.clear();
- tvListCache.get(dataType).add(list);
+ if (nvm) {
+ nvmTVListCache.get(dataType).add((NVMTVList) list);
+ } else {
+ tvListCache.get(dataType).add((TVList) list);
+ }
}
- public synchronized void release(TVList list) {
+ public synchronized void release(AbstractTVList list) {
list.clear();
- if (list instanceof BinaryTVList) {
- tvListCache.get(TSDataType.TEXT).add(list);
- } else if (list instanceof BooleanTVList) {
- tvListCache.get(TSDataType.BOOLEAN).add(list);
- } else if (list instanceof DoubleTVList) {
- tvListCache.get(TSDataType.DOUBLE).add(list);
- } else if (list instanceof FloatTVList) {
- tvListCache.get(TSDataType.FLOAT).add(list);
- } else if (list instanceof IntTVList) {
- tvListCache.get(TSDataType.INT32).add(list);
- } else if (list instanceof LongTVList) {
- tvListCache.get(TSDataType.INT64).add(list);
+ if (list instanceof NVMIntTVList) {
+ nvmTVListCache.get(TSDataType.INT32).add((NVMTVList) list);
+ } else {
+ TVList tvList = (TVList) list;
+
+ if (list instanceof BinaryTVList) {
+ tvListCache.get(TSDataType.TEXT).add(tvList);
+ } else if (list instanceof BooleanTVList) {
+ tvListCache.get(TSDataType.BOOLEAN).add(tvList);
+ } else if (list instanceof DoubleTVList) {
+ tvListCache.get(TSDataType.DOUBLE).add(tvList);
+ } else if (list instanceof FloatTVList) {
+ tvListCache.get(TSDataType.FLOAT).add(tvList);
+ } else if (list instanceof IntTVList) {
+ tvListCache.get(TSDataType.INT32).add(tvList);
+ } else if (list instanceof LongTVList) {
+ tvListCache.get(TSDataType.INT64).add(tvList);
+ }
}
}
@Override
- public int getNumberOfTVLists() {
+ public int getNumberOfTVLists(boolean nvm) {
int number = 0;
- for (Queue<TVList> queue : tvListCache.values()) {
- number += queue.size();
+ if (nvm) {
+ for (Queue<NVMTVList> queue : nvmTVListCache.values()) {
+ number += queue.size();
+ }
+ } else {
+ for (Queue<TVList> queue : tvListCache.values()) {
+ number += queue.size();
+ }
}
return number;
}
@@ -101,6 +129,7 @@ public class TVListAllocator implements TVListAllocatorMBean, IService {
public void stop() {
JMXService.deregisterMBean(mbeanName);
tvListCache.clear();
+ nvmTVListCache.clear();
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocatorMBean.java b/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocatorMBean.java
index 8dccd91..d2cdee0 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocatorMBean.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/TVListAllocatorMBean.java
@@ -20,6 +20,6 @@ package org.apache.iotdb.db.rescon;
public interface TVListAllocatorMBean {
- int getNumberOfTVLists();
+ int getNumberOfTVLists(boolean nvm);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
index 56d247d..c56c6d4 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
@@ -89,6 +89,7 @@ public class BinaryTVList extends TVList {
return cloneArray;
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -105,7 +106,7 @@ public class BinaryTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
for (Binary[] dataArray : values) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -115,7 +116,7 @@ public class BinaryTVList extends TVList {
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (Binary[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -129,17 +130,20 @@ public class BinaryTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
Binary srcV = getBinary(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getBinary(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
index 1f2d264..634b3fa 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
@@ -88,6 +88,7 @@ public class BooleanTVList extends TVList {
return cloneArray;
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -104,7 +105,7 @@ public class BooleanTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
for (boolean[] dataArray : values) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -114,7 +115,7 @@ public class BooleanTVList extends TVList {
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (boolean[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -128,17 +129,20 @@ public class BooleanTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
boolean srcV = getBoolean(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getBoolean(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
index 4efd2b4..33b76c6 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
@@ -88,6 +88,7 @@ public class DoubleTVList extends TVList {
return cloneArray;
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -104,7 +105,7 @@ public class DoubleTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
for (double[] dataArray : values) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -114,7 +115,7 @@ public class DoubleTVList extends TVList {
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (double[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -128,17 +129,20 @@ public class DoubleTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
double srcV = getDouble(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getDouble(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
index ce33e1f..b0ed801 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
@@ -88,6 +88,7 @@ public class FloatTVList extends TVList {
return cloneArray;
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -104,7 +105,7 @@ public class FloatTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
for (float[] dataArray : values) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -114,7 +115,7 @@ public class FloatTVList extends TVList {
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (float[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -128,17 +129,20 @@ public class FloatTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
float srcV = getFloat(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getFloat(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
index 8726a9e..e82ebea 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
@@ -88,6 +88,7 @@ public class IntTVList extends TVList {
return cloneArray;
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -104,7 +105,7 @@ public class IntTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
for (int[] dataArray : values) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -114,7 +115,7 @@ public class IntTVList extends TVList {
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (int[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -128,17 +129,20 @@ public class IntTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
int srcV = getInt(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getInt(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
index 6699ef4..604c973 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
@@ -88,6 +88,7 @@ public class LongTVList extends TVList {
return cloneArray;
}
+ @Override
public void sort() {
if (sortedTimestamps == null || sortedTimestamps.length < size) {
sortedTimestamps = (long[][]) PrimitiveArrayPool
@@ -104,7 +105,7 @@ public class LongTVList extends TVList {
}
@Override
- void clearValue() {
+ protected void clearValue() {
if (values != null) {
for (long[] dataArray : values) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -114,7 +115,7 @@ public class LongTVList extends TVList {
}
@Override
- void clearSortedValue() {
+ protected void clearSortedValue() {
if (sortedValues != null) {
for (long[] dataArray : sortedValues) {
PrimitiveArrayPool.getInstance().release(dataArray);
@@ -128,17 +129,20 @@ public class LongTVList extends TVList {
set(dest, sortedTimestamps[src/ARRAY_SIZE][src%ARRAY_SIZE], sortedValues[src/ARRAY_SIZE][src%ARRAY_SIZE]);
}
+ @Override
protected void set(int src, int dest) {
long srcT = getTime(src);
long srcV = getLong(src);
set(dest, srcT, srcV);
}
+ @Override
protected void setToSorted(int src, int dest) {
sortedTimestamps[dest/ARRAY_SIZE][dest% ARRAY_SIZE] = getTime(src);
sortedValues[dest/ARRAY_SIZE][dest%ARRAY_SIZE] = getLong(src);
}
+ @Override
protected void reverseRange(int lo, int hi) {
hi--;
while (lo < hi) {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
index 4e9891a..d32e7e9 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -23,31 +23,14 @@ import static org.apache.iotdb.db.rescon.PrimitiveArrayPool.ARRAY_SIZE;
import java.util.ArrayList;
import java.util.List;
+import org.apache.iotdb.db.nvm.datastructure.AbstractTVList;
import org.apache.iotdb.db.rescon.PrimitiveArrayPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.utils.Binary;
@SuppressWarnings("unused")
-public abstract class TVList {
-
- private static final String ERR_DATATYPE_NOT_CONSISTENT = "DataType not consistent";
-
- protected static final int SMALL_ARRAY_LENGTH = 32;
+public abstract class TVList extends AbstractTVList {
protected List<long[]> timestamps;
- protected int size;
-
- protected long[][] sortedTimestamps;
- protected boolean sorted = true;
-
- /**
- * this field is effective only in the Tvlist in a RealOnlyMemChunk.
- */
- private long timeOffset = Long.MIN_VALUE;
-
- protected long pivotTime;
-
- protected long minTime;
public TVList() {
timestamps = new ArrayList<>();
@@ -55,10 +38,7 @@ public abstract class TVList {
minTime = Long.MIN_VALUE;
}
- public int size() {
- return size;
- }
-
+ @Override
public long getTime(int index) {
if (index >= size) {
throw new ArrayIndexOutOfBoundsException(index);
@@ -68,102 +48,12 @@ public abstract class TVList {
return timestamps.get(arrayIndex)[elementIndex];
}
- public void putLong(long time, long value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putInt(long time, int value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putFloat(long time, float value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putDouble(long time, double value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putBinary(long time, Binary value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putBoolean(long time, boolean value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putLongs(long[] time, long[] value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putInts(long[] time, int[] value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putFloats(long[] time, float[] value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putDoubles(long[] time, double[] value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putBinaries(long[] time, Binary[] value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public void putBooleans(long[] time, boolean[] value) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public long getLong(int index) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public int getInt(int index) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public float getFloat(int index) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public double getDouble(int index) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public Binary getBinary(int index) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public boolean getBoolean(int index) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
- public abstract void sort();
-
- public long getMinTime() {
- return minTime;
- }
-
- protected abstract void set(int src, int dest);
-
- protected abstract void setFromSorted(int src, int dest);
-
- protected abstract void setToSorted(int src, int dest);
-
- protected abstract void reverseRange(int lo, int hi);
-
- protected abstract void expandValues();
-
- public abstract TVList clone();
-
- protected abstract void releaseLastValueArray();
-
+ @Override
protected void releaseLastTimeArray() {
PrimitiveArrayPool.getInstance().release(timestamps.remove(timestamps.size() - 1));
}
+ @Override
public void delete(long upperBound) {
int newSize = 0;
minTime = Long.MAX_VALUE;
@@ -186,27 +76,18 @@ public abstract class TVList {
}
}
- protected void cloneAs(TVList cloneList) {
+ @Override
+ protected void cloneAs(AbstractTVList abstractCloneList) {
+ TVList cloneList = (TVList) abstractCloneList;
for (long[] timestampArray : timestamps) {
- cloneList.timestamps.add(cloneTime(timestampArray));
+ cloneList.timestamps.add((long[]) cloneTime(timestampArray));
}
cloneList.size = size;
cloneList.sorted = sorted;
cloneList.minTime = minTime;
}
- public void clear() {
- size = 0;
- timeOffset = Long.MIN_VALUE;
- sorted = true;
- minTime = Long.MIN_VALUE;
- clearTime();
- clearSortedTime();
-
- clearValue();
- clearSortedValue();
- }
-
+ @Override
protected void clearTime() {
if (timestamps != null) {
for (long[] dataArray : timestamps) {
@@ -216,6 +97,7 @@ public abstract class TVList {
}
}
+ @Override
protected void clearSortedTime() {
if (sortedTimestamps != null) {
for (long[] dataArray : sortedTimestamps) {
@@ -225,62 +107,23 @@ public abstract class TVList {
}
}
- abstract void clearValue();
-
- abstract void clearSortedValue();
-
+ @Override
protected void checkExpansion() {
if ((size % ARRAY_SIZE) == 0) {
expandValues();
- timestamps.add((long[]) PrimitiveArrayPool.getInstance().getPrimitiveDataListByType(TSDataType.INT64));
+ timestamps.add((long[]) PrimitiveArrayPool.getInstance().getPrimitiveDataListByType(
+ TSDataType.INT64));
}
}
- protected long[] cloneTime(long[] array) {
+ @Override
+ protected Object cloneTime(Object object) {
+ long[] array = (long[]) object;
long[] cloneArray = new long[array.length];
System.arraycopy(array, 0, cloneArray, 0, array.length);
return cloneArray;
}
- protected void sort(int lo, int hi) {
- if (sorted) {
- return;
- }
- if (lo == hi) {
- return;
- }
- if (hi - lo <= SMALL_ARRAY_LENGTH) {
- int initRunLen = countRunAndMakeAscending(lo, hi);
- binarySort(lo, hi, lo + initRunLen);
- return;
- }
- int mid = (lo + hi) >>> 1;
- sort(lo, mid);
- sort(mid, hi);
- merge(lo, mid, hi);
- }
-
- protected int countRunAndMakeAscending(int lo, int hi) {
- assert lo < hi;
- int runHi = lo + 1;
- if (runHi == hi) {
- return 1;
- }
-
- // Find end of run, and reverse range if descending
- if (getTime(runHi++) < getTime(lo)) { // Descending
- while (runHi < hi && getTime(runHi) < getTime(runHi - 1)) {
- runHi++;
- }
- reverseRange(lo, runHi);
- } else { // Ascending
- while (runHi < hi &&getTime(runHi) >= getTime(runHi - 1))
- runHi++;
- }
-
- return runHi - lo;
- }
-
public static TVList newList(TSDataType dataType) {
switch (dataType) {
case TEXT:
@@ -298,137 +141,4 @@ public abstract class TVList {
}
return null;
}
-
- /**
- * this field is effective only in the Tvlist in a RealOnlyMemChunk.
- * @return
- */
- public long getTimeOffset() {
- return timeOffset;
- }
-
- public void setTimeOffset(long timeOffset) {
- this.timeOffset = timeOffset;
- }
-
- protected int compare(int idx1, int idx2) {
- long t1 = getTime(idx1);
- long t2 = getTime(idx2);
- return Long.compare(t1, t2);
- }
-
- protected abstract void saveAsPivot(int pos);
-
- protected abstract void setPivotTo(int pos);
-
- /**
- * From TimSort.java
- */
- protected void binarySort(int lo, int hi, int start) {
- assert lo <= start && start <= hi;
- if (start == lo)
- start++;
- for ( ; start < hi; start++) {
-
- saveAsPivot(start);
- // Set left (and right) to the index where a[start] (pivot) belongs
- int left = lo;
- int right = start;
- assert left <= right;
- /*
- * Invariants:
- * pivot >= all in [lo, left).
- * pivot < all in [right, start).
- */
- while (left < right) {
- int mid = (left + right) >>> 1;
- if (compare(start, mid) < 0)
- right = mid;
- else
- left = mid + 1;
- }
- assert left == right;
-
- /*
- * The invariants still hold: pivot >= all in [lo, left) and
- * pivot < all in [left, start), so pivot belongs at left. Note
- * that if there are elements equal to pivot, left points to the
- * first slot after them -- that's why this sort is stable.
- * Slide elements over to make room for pivot.
- */
- int n = start - left; // The number of elements to move
- for (int i = n; i >= 1; i--) {
- set(left + i - 1, left + i);
- }
- setPivotTo(left);
- }
- for (int i = lo; i < hi; i++) {
- setToSorted(i, i);
- }
- }
-
- protected void merge(int lo, int mid, int hi) {
- // end of sorting buffer
- int tmpIdx = 0;
-
- // start of unmerged parts of each sequence
- int leftIdx = lo;
- int rightIdx = mid;
-
- // copy the minimum elements to sorting buffer until one sequence is exhausted
- int endSide = 0;
- while (endSide == 0) {
- if (compare(leftIdx, rightIdx) <= 0) {
- setToSorted(leftIdx, lo + tmpIdx);
- tmpIdx ++;
- leftIdx ++;
- if (leftIdx == mid) {
- endSide = 1;
- }
- } else {
- setToSorted(rightIdx, lo + tmpIdx);
- tmpIdx ++;
- rightIdx ++;
- if (rightIdx == hi) {
- endSide = 2;
- }
- }
- }
-
- // copy the remaining elements of another sequence
- int start;
- int end;
- if (endSide == 1) {
- start = rightIdx;
- end = hi;
- } else {
- start = leftIdx;
- end = mid;
- }
- for (; start < end; start++) {
- setToSorted(start, lo + tmpIdx);
- tmpIdx ++;
- }
-
- // copy from sorting buffer to the original arrays so that they can be further sorted
- // potential speed up: change the place of sorting buffer and origin data between merge
- // iterations
- for (int i = lo; i < hi; i++) {
- setFromSorted(i, i);
- }
- }
-
- void updateMinTimeAndSorted(long[] time) {
- int length = time.length;
- long inPutMinTime = Long.MAX_VALUE;
- boolean inputSorted = true;
- for (int i = 0; i < length; i++) {
- inPutMinTime = inPutMinTime <= time[i] ? inPutMinTime : time[i];
- if (inputSorted && i < length - 1 && time[i] > time[i+1]) {
- inputSorted = false;
- }
- }
- minTime = inPutMinTime < minTime ? inPutMinTime : minTime;
- sorted = sorted && inputSorted && (size == 0 || inPutMinTime >= getTime(size - 1));
- }
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java b/server/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java
index 0c617bc..ea51b07 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/memtable/MemTablePoolTest.java
@@ -47,7 +47,7 @@ public class MemTablePoolTest {
public void testGetAndRelease() {
long time = System.currentTimeMillis();
for (int i = 0; i < 10; i++) {
- IMemTable memTable = MemTablePool.getInstance().getAvailableMemTable("test case");
+ IMemTable memTable = MemTablePool.getInstance().getAvailableMemTable("test case", false);
memTables.add(memTable);
}
time -= System.currentTimeMillis();
@@ -86,7 +86,7 @@ public class MemTablePoolTest {
continue;
}
memTables.remove(memTable);
- MemTablePool.getInstance().putBack(memTable, "test case");
+ MemTablePool.getInstance().putBack(memTable, "test case", false);
}
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/PerfTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/PerfTest.java
new file mode 100644
index 0000000..c42df49
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/PerfTest.java
@@ -0,0 +1,148 @@
+package org.apache.iotdb.db.engine.storagegroup;
+
+import static junit.framework.TestCase.assertTrue;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.MetadataManagerHelper;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
+import org.apache.iotdb.db.engine.version.SysTimeVersionController;
+import org.apache.iotdb.db.exception.StartupException;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.db.utils.SchemaUtils;
+import org.apache.iotdb.db.utils.TimeValuePair;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.write.record.TSRecord;
+import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
+import org.junit.Test;
+
+public class PerfTest {
+
+ private static TsFileProcessor processor;
+ private static String storageGroup = "storage_group1";
+ private static String filePath = "data/testUnsealedTsFileProcessor.tsfile";
+ private static String deviceId = "root.vehicle.d0";
+ private static String measurementId = "s0";
+ private static TSDataType dataType = TSDataType.INT32;
+ private static Map<String, String> props = Collections.emptyMap();
+ private static QueryContext context;
+
+ private static TsFileProcessor processor2;
+ private static String storageGroup2 = "storage_group2";
+ private static String filePath2 = "data/testUnsealedTsFileProcessor.tsfile2";
+ private static String deviceId2 = "root.vehicle.d2";
+ private static String measurementId2 = "s2";
+ private static TSDataType dataType2 = TSDataType.INT32;
+ private static Map<String, String> props2 = Collections.emptyMap();
+ private static QueryContext context2;
+
+ public static void main(String[] args)
+ throws IOException, StartupException, StorageEngineException, QueryProcessException {
+ args = new String[]{"data/testUnsealedTsFileProcessor.tsfile", "data/testUnsealedTsFileProcessor.tsfile2", "data/nvmFile", "true", "10000"};
+ filePath = args[0];
+ filePath2 = args[1];
+ String nvmPath = args[2];
+ IoTDBDescriptor.getInstance().getConfig().setEnableWal(Boolean.parseBoolean(args[3]));
+ IoTDBDescriptor.getInstance().getConfig().setFlushWalThreshold(Integer.parseInt(args[4]));
+
+ int recordNum = 100000;
+ long time;
+
+ MetadataManagerHelper.initMetadata();
+ EnvironmentUtils.envSetUp();
+ context = EnvironmentUtils.TEST_QUERY_CONTEXT;
+ context2 = EnvironmentUtils.TEST_QUERY_CONTEXT;
+ NVMSpaceManager.init(nvmPath);
+
+ processor = new TsFileProcessor(storageGroup, SystemFileFactory.INSTANCE.getFile(filePath),
+ SchemaUtils.constructSchema(deviceId), SysTimeVersionController.INSTANCE, x -> {
+ },
+ () -> true, true);
+
+ processor2 = new TsFileProcessor(storageGroup2, SystemFileFactory.INSTANCE.getFile(filePath2),
+ SchemaUtils.constructSchema(deviceId2), SysTimeVersionController.INSTANCE, x -> {
+ },
+ () -> true, true);
+ processor2.setUseNVM(true);
+
+ Pair<ReadOnlyMemChunk, List<ChunkMetaData>> pair = processor
+ .query(deviceId, measurementId, dataType, props, context);
+ ReadOnlyMemChunk left = pair.left;
+ List<ChunkMetaData> right = pair.right;
+ assertTrue(left.isEmpty());
+ assertEquals(0, right.size());
+
+ Pair<ReadOnlyMemChunk, List<ChunkMetaData>> pair2 = processor2
+ .query(deviceId2, measurementId2, dataType2, props2, context2);
+ ReadOnlyMemChunk left2 = pair.left;
+ List<ChunkMetaData> right2 = pair.right;
+ assertTrue(left2.isEmpty());
+ assertEquals(0, right2.size());
+
+ System.out.println("insert start");
+
+ time = System.currentTimeMillis();
+ for (int i = 1; i <= recordNum; i++) {
+ TSRecord record = new TSRecord(i, deviceId);
+ record.addTuple(DataPoint.getDataPoint(dataType, measurementId, String.valueOf(i)));
+ processor.insert(new InsertPlan(record));
+ }
+ System.out.println("normal:" + (System.currentTimeMillis() - time));
+
+ time = System.currentTimeMillis();
+ for (int i = 1; i <= recordNum; i++) {
+ TSRecord record = new TSRecord(i, deviceId2);
+ record.addTuple(DataPoint.getDataPoint(dataType2, measurementId2, String.valueOf(i)));
+ processor2.insert(new InsertPlan(record));
+ }
+ System.out.println("nvm:" + (System.currentTimeMillis() - time));
+
+ // query data in memory
+ pair = processor.query(deviceId, measurementId, dataType, props, context);
+ left = pair.left;
+ assertFalse(left.isEmpty());
+ int num = 1;
+ Iterator<TimeValuePair> iterator = left.getIterator();
+ for (; num <= 100; num++) {
+ iterator.hasNext();
+ TimeValuePair timeValuePair = iterator.next();
+ assertEquals(num, timeValuePair.getTimestamp());
+ assertEquals(num, timeValuePair.getValue().getInt());
+ }
+
+ pair2 = processor2.query(deviceId2, measurementId2, dataType2, props2, context2);
+ left2 = pair2.left;
+ assertFalse(left2.isEmpty());
+ int num2 = 1;
+ Iterator<TimeValuePair> iterator2 = left2.getIterator();
+ for (; num2 <= 100; num2++) {
+ iterator2.hasNext();
+ TimeValuePair timeValuePair2 = iterator2.next();
+ assertEquals(num2, timeValuePair2.getTimestamp());
+ assertEquals(num2, timeValuePair2.getValue().getInt());
+ }
+
+
+ EnvironmentUtils.cleanEnv();
+ new File(filePath).deleteOnExit();
+ new File(filePath2).deleteOnExit();
+ new File(nvmPath).deleteOnExit();
+ System.exit(0);
+ System.out.println("exit");
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java
index 5e8d60e..fad7bc1 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.engine.version.SysTimeVersionController;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.nvm.space.NVMSpaceManager;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -61,11 +62,21 @@ public class TsFileProcessorTest {
private Map<String, String> props = Collections.emptyMap();
private QueryContext context;
+ private TsFileProcessor processor2;
+ private String storageGroup2 = "storage_group2";
+ private String filePath2 = "data/testUnsealedTsFileProcessor.tsfile2";
+ private String deviceId2 = "root.vehicle.d2";
+ private String measurementId2 = "s2";
+ private TSDataType dataType2 = TSDataType.INT32;
+ private Map<String, String> props2 = Collections.emptyMap();
+ private QueryContext context2;
+
@Before
public void setUp() throws Exception {
MetadataManagerHelper.initMetadata();
EnvironmentUtils.envSetUp();
context = EnvironmentUtils.TEST_QUERY_CONTEXT;
+ context2 = EnvironmentUtils.TEST_QUERY_CONTEXT;
}
@After
diff --git a/server/src/test/java/org/apache/iotdb/db/utils/datastructure/LongTVListTest.java b/server/src/test/java/org/apache/iotdb/db/utils/datastructure/LongTVListTest.java
index c4f3a40..a55de1b 100644
--- a/server/src/test/java/org/apache/iotdb/db/utils/datastructure/LongTVListTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/utils/datastructure/LongTVListTest.java
@@ -36,7 +36,7 @@ public class LongTVListTest {
tvList.putLong(i, i);
}
tvList.sort();
- for (long i = 0; i < tvList.size; i++) {
+ for (long i = 0; i < tvList.size(); i++) {
Assert.assertEquals(i, tvList.getLong((int)i));
Assert.assertEquals(i, tvList.getTime((int)i));
}
@@ -49,7 +49,7 @@ public class LongTVListTest {
tvList.putLong(i, i);
}
tvList.sort();
- for (long i = 0; i < tvList.size; i++) {
+ for (long i = 0; i < tvList.size(); i++) {
Assert.assertEquals(i, tvList.getLong((int)i));
Assert.assertEquals(i, tvList.getTime((int)i));
}
@@ -68,7 +68,7 @@ public class LongTVListTest {
}
tvList.sort();
inputs.sort(TimeValuePair::compareTo);
- for (long i = 0; i < tvList.size; i++) {
+ for (long i = 0; i < tvList.size(); i++) {
Assert.assertEquals(inputs.get((int)i).getTimestamp(), tvList.getTime((int)i));
Assert.assertEquals(inputs.get((int)i).getValue().getLong(), tvList.getLong((int)i));
}
@@ -83,7 +83,7 @@ public class LongTVListTest {
tvList.putLong(i, i);
}
tvList.sort();
- for (int i = 0; i < tvList.size; i++) {
+ for (int i = 0; i < tvList.size(); i++) {
tvList.getLong(i);
tvList.getTime(i);
}