This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 5968100a391 [To dev/1.3]Memtable enhance fix (#15040)
5968100a391 is described below
commit 5968100a3911c7113abaea029dfb4dba85fea391
Author: shizy <[email protected]>
AuthorDate: Fri Mar 7 23:04:06 2025 +0800
[To dev/1.3]Memtable enhance fix (#15040)
* perf: improve seq inserting
* rename MergeSortTvListIterator to MergeSortTVListIterator
* fix: concurrent indices modification during query sort and flush sort
* revert: iterate rows in aligned memChunk
* fix: testMemAlignedChunkLoader test
* fix: no need to synchronize list since sort is already an synchronized
method
* fix: optimize mininum time update in TVList
* perf: special case for merge sort iterator when no handover occurs
---
.../dataregion/memtable/AbstractMemTable.java | 9 -
.../memtable/AlignedReadOnlyMemChunk.java | 314 +++++++--------------
.../memtable/AlignedWritableMemChunk.java | 134 +++------
.../memtable/AlignedWritableMemChunkGroup.java | 5 -
.../dataregion/memtable/IMemTable.java | 2 -
.../memtable/IWritableMemChunkGroup.java | 2 -
.../dataregion/memtable/ReadOnlyMemChunk.java | 16 +-
.../dataregion/memtable/TsFileProcessor.java | 12 +-
.../dataregion/memtable/WritableMemChunk.java | 124 ++++----
.../dataregion/memtable/WritableMemChunkGroup.java | 8 -
.../read/reader/chunk/MemAlignedChunkReader.java | 186 ++++--------
.../read/reader/chunk/MemChunkReader.java | 4 +-
.../read/reader/chunk/MemPageReader.java | 6 +-
.../db/utils/datastructure/AlignedTVList.java | 55 ++--
.../iotdb/db/utils/datastructure/BinaryTVList.java | 2 +-
.../db/utils/datastructure/BooleanTVList.java | 2 +-
.../iotdb/db/utils/datastructure/DoubleTVList.java | 2 +-
.../iotdb/db/utils/datastructure/FloatTVList.java | 2 +-
.../iotdb/db/utils/datastructure/IntTVList.java | 2 +-
.../iotdb/db/utils/datastructure/LongTVList.java | 2 +-
.../MergeSortAlignedTVListIterator.java | 172 ++++-------
...tIterator.java => MergeSortTVListIterator.java} | 90 +++---
.../utils/datastructure/PageColumnAccessInfo.java | 52 ----
.../iotdb/db/utils/datastructure/TVList.java | 4 +-
.../read/reader/chunk/MemChunkLoaderTest.java | 26 +-
25 files changed, 435 insertions(+), 798 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
index 6962b8bcafe..8d07fa50c8c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
@@ -385,15 +385,6 @@ public abstract class AbstractMemTable implements
IMemTable {
return !memChunkGroup.contains(measurement);
}
- @Override
- public long getMeasurementSize(IDeviceID deviceId, String measurement) {
- IWritableMemChunkGroup memChunkGroup = memTableMap.get(deviceId);
- if (null == memChunkGroup) {
- return 0;
- }
- return memChunkGroup.getMeasurementSize(measurement);
- }
-
@Override
public IWritableMemChunk getWritableMemChunk(IDeviceID deviceId, String
measurement) {
IWritableMemChunkGroup memChunkGroup = memTableMap.get(deviceId);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
index e61b65db7f9..7c6264f1097 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
@@ -23,7 +23,6 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import
org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemAlignedChunkLoader;
import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.MergeSortAlignedTVListIterator;
-import org.apache.iotdb.db.utils.datastructure.PageColumnAccessInfo;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.tsfile.block.column.ColumnBuilder;
@@ -141,33 +140,30 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
alignedTvLists, dataTypes, columnIndexList, floatPrecision,
encodingList);
int[] alignedTvListOffsets =
timeValuePairIterator.getAlignedTVListOffsets();
- // iterate to build column access info and split pages
- int pointsInPage = 0;
- long[] time = new long[MAX_NUMBER_OF_POINTS_IN_PAGE];
- PageColumnAccessInfo[] pageColumnAccessInfo = new
PageColumnAccessInfo[dataTypes.size()];
- for (int i = 0; i < pageColumnAccessInfo.length; i++) {
- pageColumnAccessInfo[i] = new
PageColumnAccessInfo(MAX_NUMBER_OF_POINTS_IN_PAGE);
- }
-
+ int pointsInChunk = 0;
int[] timeDeleteCursor = new int[] {0};
List<int[]> valueColumnDeleteCursor = new ArrayList<>();
if (valueColumnsDeletionList != null) {
valueColumnsDeletionList.forEach(x -> valueColumnDeleteCursor.add(new
int[] {0}));
}
+ Statistics<? extends Serializable> pageTimeStats = null;
while (timeValuePairIterator.hasNextTimeValuePair()) {
- long timestamp = timeValuePairIterator.getTime();
+ TimeValuePair tvPair = timeValuePairIterator.currentTimeValuePair();
+ TsPrimitiveType[] values = tvPair.getValue().getVector();
// ignore all-null row
- BitMap bitMap = timeValuePairIterator.getBitmap();
- if (valueColumnsDeletionList != null) {
- for (int columnIndex = 0; columnIndex < dataTypes.size();
columnIndex++) {
- if (isPointDeleted(
- timestamp,
- valueColumnsDeletionList.get(columnIndex),
- valueColumnDeleteCursor.get(columnIndex))) {
- bitMap.mark(columnIndex);
- }
+ BitMap bitMap = new BitMap(dataTypes.size());
+ for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++)
{
+ if (values[columnIndex] == null) {
+ bitMap.mark(columnIndex);
+ } else if (valueColumnsDeletionList != null
+ && isPointDeleted(
+ tvPair.getTimestamp(),
+ valueColumnsDeletionList.get(columnIndex),
+ valueColumnDeleteCursor.get(columnIndex))) {
+ values[columnIndex] = null;
+ bitMap.mark(columnIndex);
}
}
if (bitMap.isAllMarked()) {
@@ -176,67 +172,93 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
}
// create pageTimeStatistics and pageValueStatistics for new page
- if (pointsInPage == 0) {
+ if (pointsInChunk % MAX_NUMBER_OF_POINTS_IN_PAGE == 0) {
Statistics<? extends Serializable> pageTimeStatistics =
Statistics.getStatsByType(TSDataType.VECTOR);
+ pageTimeStatistics.setEmpty(false);
timeStatisticsList.add(pageTimeStatistics);
+ pageTimeStats = timeStatisticsList.get(timeStatisticsList.size() - 1);
+
Statistics<? extends Serializable>[] pageValueStatistics =
new Statistics[valueChunkNames.size()];
+ Arrays.fill(pageValueStatistics, null);
valueStatisticsList.add(pageValueStatistics);
-
pageOffsetsList.add(Arrays.copyOf(alignedTvListOffsets,
alignedTvListOffsets.length));
}
- // prepare column access info for current page
- int[][] columnAccessInfo = timeValuePairIterator.getColumnAccessInfo();
- time[pointsInPage] = timeValuePairIterator.getTime();
- for (int i = 0; i < dataTypes.size(); i++) {
- pageColumnAccessInfo[i].add(columnAccessInfo[i]);
- }
- timeValuePairIterator.step();
- pointsInPage++;
+ pageTimeStats.update(tvPair.getTimestamp());
+ chunkTimeStatistics.update(tvPair.getTimestamp());
- if (pointsInPage == MAX_NUMBER_OF_POINTS_IN_PAGE) {
- Statistics<? extends Serializable> pageTimeStatistics =
- timeStatisticsList.get(timeStatisticsList.size() - 1);
- Statistics<? extends Serializable>[] pageValueStatistics =
- valueStatisticsList.get(valueStatisticsList.size() - 1);
-
- // update page time & value statistics
- updateTimeStatistics(
- time, MAX_NUMBER_OF_POINTS_IN_PAGE, chunkTimeStatistics,
pageTimeStatistics);
- updateValueStatistics(
- time,
- pageColumnAccessInfo,
- chunkValueStatistics,
- pageValueStatistics,
- timeValuePairIterator);
-
- // reset
- for (PageColumnAccessInfo p : pageColumnAccessInfo) {
- p.reset();
+ Statistics<? extends Serializable>[] pageValuesStats =
+ valueStatisticsList.get(valueStatisticsList.size() - 1);
+ TsPrimitiveType[] primitiveValues = tvPair.getValue().getVector();
+ for (int column = 0; column < primitiveValues.length; column++) {
+ if (primitiveValues[column] == null) {
+ continue;
}
- chunkTimeStatistics.setEmpty(false);
- pointsInPage = 0;
- }
- }
- if (pointsInPage > 0) {
- Statistics<? extends Serializable> pageTimeStatistics =
- timeStatisticsList.get(timeStatisticsList.size() - 1);
- Statistics<? extends Serializable>[] pageValueStatistics =
- valueStatisticsList.get(valueStatisticsList.size() - 1);
+ if (pageValuesStats[column] == null) {
+ Statistics<? extends Serializable> valueStatistics =
+ Statistics.getStatsByType(dataTypes.get(column));
+ pageValuesStats[column] = valueStatistics;
+ }
+ if (chunkValueStatistics[column] == null) {
+ Statistics<? extends Serializable> chunkValueStats =
+ Statistics.getStatsByType(dataTypes.get(column));
+ chunkValueStatistics[column] = chunkValueStats;
+ }
- updateTimeStatistics(time, pointsInPage, chunkTimeStatistics,
pageTimeStatistics);
- updateValueStatistics(
- time,
- pageColumnAccessInfo,
- chunkValueStatistics,
- pageValueStatistics,
- timeValuePairIterator);
- chunkTimeStatistics.setEmpty(false);
+ switch (dataTypes.get(column)) {
+ case BOOLEAN:
+ pageValuesStats[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getBoolean());
+ chunkValueStatistics[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getBoolean());
+ break;
+ case INT32:
+ case DATE:
+ pageValuesStats[column].update(tvPair.getTimestamp(),
primitiveValues[column].getInt());
+ chunkValueStatistics[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getInt());
+ break;
+ case INT64:
+ case TIMESTAMP:
+ pageValuesStats[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getLong());
+ chunkValueStatistics[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getLong());
+ break;
+ case FLOAT:
+ pageValuesStats[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getFloat());
+ chunkValueStatistics[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getFloat());
+ break;
+ case DOUBLE:
+ pageValuesStats[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getDouble());
+ chunkValueStatistics[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getDouble());
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ pageValuesStats[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getBinary());
+ chunkValueStatistics[column].update(
+ tvPair.getTimestamp(), primitiveValues[column].getBinary());
+ break;
+ default:
+ throw new UnSupportedDataTypeException(
+ String.format("Data type %s is not supported.",
dataTypes.get(column)));
+ }
+ }
+ timeValuePairIterator.step();
+ pointsInChunk++;
}
+
pageOffsetsList.add(Arrays.copyOf(alignedTvListOffsets,
alignedTvListOffsets.length));
+ chunkTimeStatistics.setEmpty(pointsInChunk == 0);
// aligned chunk meta
List<IChunkMetadata> valueChunkMetadataList = new ArrayList<>();
@@ -297,15 +319,6 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
}
}
- private boolean isAllColumnNull(TsPrimitiveType[] primitiveValues) {
- for (TsPrimitiveType primitiveValue : primitiveValues) {
- if (primitiveValue != null) {
- return false;
- }
- }
- return true;
- }
-
private void writeValidValuesIntoTsBlock(TsBlockBuilder builder) throws
IOException {
List<AlignedTVList> alignedTvLists =
alignedTvListQueryMap.keySet().stream()
@@ -323,26 +336,27 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
while (timeValuePairIterator.hasNextTimeValuePair()) {
TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
+ TsPrimitiveType[] values = tvPair.getValue().getVector();
- TsPrimitiveType[] primitiveValues = tvPair.getValue().getVector();
- if (valueColumnsDeletionList != null) {
- for (int columnIndex = 0; columnIndex < primitiveValues.length;
columnIndex++) {
- if (isPointDeleted(
- tvPair.getTimestamp(),
- valueColumnsDeletionList.get(columnIndex),
- valueColumnDeleteCursor.get(columnIndex))) {
- primitiveValues[columnIndex] = null;
- }
+ BitMap bitMap = new BitMap(values.length);
+ for (int columnIndex = 0; columnIndex < values.length; columnIndex++) {
+ if (values[columnIndex] == null) {
+ bitMap.mark(columnIndex);
+ } else if (valueColumnsDeletionList != null
+ && isPointDeleted(
+ tvPair.getTimestamp(),
+ valueColumnsDeletionList.get(columnIndex),
+ valueColumnDeleteCursor.get(columnIndex))) {
+ values[columnIndex] = null;
+ bitMap.mark(columnIndex);
}
}
- if (isAllColumnNull(primitiveValues)) {
- timeValuePairIterator.step();
+ if (bitMap.isAllMarked()) {
continue;
}
builder.getTimeColumnBuilder().writeLong(tvPair.getTimestamp());
// value columns
- TsPrimitiveType[] values = tvPair.getValue().getVector();
for (int columnIndex = 0; columnIndex < values.length; columnIndex++) {
if (values[columnIndex] == null) {
builder.getColumnBuilder(columnIndex).appendNull();
@@ -416,132 +430,4 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
public MergeSortAlignedTVListIterator getMergeSortAlignedTVListIterator() {
return timeValuePairIterator;
}
-
- private void updateTimeStatistics(
- long[] time,
- int count,
- Statistics<? extends Serializable> chunkTimeStatistics,
- Statistics<? extends Serializable> pageTimeStatistics) {
- for (int index = 0; index < count; index++) {
- chunkTimeStatistics.update(time[index]);
- pageTimeStatistics.update(time[index]);
- }
- pageTimeStatistics.setEmpty(count == 0);
- }
-
- private void createValueStatisticsIfNotExists(
- Statistics<? extends Serializable>[] chunkValueStatistics,
- Statistics<? extends Serializable>[] pageValueStatistics,
- List<TSDataType> dataTypes,
- int columnIndex) {
- if (pageValueStatistics[columnIndex] == null) {
- Statistics<? extends Serializable> valueStatistics =
- Statistics.getStatsByType(dataTypes.get(columnIndex));
- pageValueStatistics[columnIndex] = valueStatistics;
- }
- if (chunkValueStatistics[columnIndex] == null) {
- Statistics<? extends Serializable> chunkValueStats =
- Statistics.getStatsByType(dataTypes.get(columnIndex));
- chunkValueStatistics[columnIndex] = chunkValueStats;
- }
- }
-
- private void updateValueStatistics(
- long[] time,
- PageColumnAccessInfo[] columnAccessInfo,
- Statistics<? extends Serializable>[] chunkValueStatistics,
- Statistics<? extends Serializable>[] pageValueStatistics,
- MergeSortAlignedTVListIterator timeValuePairIterator) {
- // update value statistics
- for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) {
- PageColumnAccessInfo pageAccessInfo = columnAccessInfo[columnIndex];
- switch (dataTypes.get(columnIndex)) {
- case BOOLEAN:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value != null) {
- createValueStatisticsIfNotExists(
- pageValueStatistics, chunkValueStatistics, dataTypes,
columnIndex);
- pageValueStatistics[columnIndex].update(time[index],
value.getBoolean());
- chunkValueStatistics[columnIndex].update(time[index],
value.getBoolean());
- }
- }
- break;
- case INT32:
- case DATE:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value != null) {
- createValueStatisticsIfNotExists(
- pageValueStatistics, chunkValueStatistics, dataTypes,
columnIndex);
- pageValueStatistics[columnIndex].update(time[index],
value.getInt());
- chunkValueStatistics[columnIndex].update(time[index],
value.getInt());
- }
- }
- break;
- case INT64:
- case TIMESTAMP:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value != null) {
- createValueStatisticsIfNotExists(
- pageValueStatistics, chunkValueStatistics, dataTypes,
columnIndex);
- pageValueStatistics[columnIndex].update(time[index],
value.getLong());
- chunkValueStatistics[columnIndex].update(time[index],
value.getLong());
- }
- }
- break;
- case FLOAT:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value != null) {
- createValueStatisticsIfNotExists(
- pageValueStatistics, chunkValueStatistics, dataTypes,
columnIndex);
- pageValueStatistics[columnIndex].update(time[index],
value.getFloat());
- chunkValueStatistics[columnIndex].update(time[index],
value.getFloat());
- }
- }
- break;
- case DOUBLE:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value != null) {
- createValueStatisticsIfNotExists(
- pageValueStatistics, chunkValueStatistics, dataTypes,
columnIndex);
- pageValueStatistics[columnIndex].update(time[index],
value.getDouble());
- chunkValueStatistics[columnIndex].update(time[index],
value.getDouble());
- }
- }
- break;
- case TEXT:
- case BLOB:
- case STRING:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value != null) {
- createValueStatisticsIfNotExists(
- pageValueStatistics, chunkValueStatistics, dataTypes,
columnIndex);
- pageValueStatistics[columnIndex].update(time[index],
value.getBinary());
- chunkValueStatistics[columnIndex].update(time[index],
value.getBinary());
- }
- }
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format("Data type %s is not supported.",
dataTypes.get(columnIndex)));
- }
- }
- }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
index 8329f528fd2..00e6240723c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
@@ -28,11 +28,11 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferVie
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils;
import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.MergeSortAlignedTVListIterator;
-import org.apache.iotdb.db.utils.datastructure.PageColumnAccessInfo;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
@@ -702,85 +702,6 @@ public class AlignedWritableMemChunk implements
IWritableMemChunk {
}
}
- private void writePageValuesIntoWriter(
- IChunkWriter chunkWriter,
- long[] times,
- PageColumnAccessInfo[] pageColumnAccessInfo,
- MergeSortAlignedTVListIterator timeValuePairIterator) {
- AlignedChunkWriterImpl alignedChunkWriter = (AlignedChunkWriterImpl)
chunkWriter;
-
- // update value statistics
- for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) {
- ValueChunkWriter valueChunkWriter =
- alignedChunkWriter.getValueChunkWriterByIndex(columnIndex);
- PageColumnAccessInfo pageAccessInfo = pageColumnAccessInfo[columnIndex];
- switch (dataTypes.get(columnIndex)) {
- case BOOLEAN:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- valueChunkWriter.write(
- times[index], value != null && value.getBoolean(), value ==
null);
- }
- break;
- case INT32:
- case DATE:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- valueChunkWriter.write(times[index], value == null ? 0 :
value.getInt(), value == null);
- }
- break;
- case INT64:
- case TIMESTAMP:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- valueChunkWriter.write(
- times[index], value == null ? 0L : value.getLong(), value ==
null);
- }
- break;
- case FLOAT:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- valueChunkWriter.write(
- times[index], value == null ? 0f : value.getFloat(), value ==
null);
- }
- break;
- case DOUBLE:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- valueChunkWriter.write(
- times[index], value == null ? 0d : value.getDouble(), value ==
null);
- }
- break;
- case TEXT:
- case BLOB:
- case STRING:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- valueChunkWriter.write(
- times[index],
- value == null ? Binary.EMPTY_VALUE : value.getBinary(),
- value == null);
- }
- break;
- default:
- throw new UnSupportedDataTypeException(
- String.format("Data type %s is not supported.",
dataTypes.get(columnIndex)));
- }
- }
- }
-
@Override
public synchronized void encode(BlockingQueue<Object> ioTaskQueue) {
if (TVLIST_SORT_THRESHOLD == 0) {
@@ -799,30 +720,49 @@ public class AlignedWritableMemChunk implements
IWritableMemChunk {
int pointNumInChunk = 0;
long[] times = new long[MAX_NUMBER_OF_POINTS_IN_PAGE];
- PageColumnAccessInfo[] pageColumnAccessInfo = new
PageColumnAccessInfo[dataTypes.size()];
- for (int i = 0; i < pageColumnAccessInfo.length; i++) {
- pageColumnAccessInfo[i] = new
PageColumnAccessInfo(MAX_NUMBER_OF_POINTS_IN_PAGE);
- }
-
while (timeValuePairIterator.hasNextTimeValuePair()) {
- // prepare column access info for current page
- int[][] accessInfo = timeValuePairIterator.getColumnAccessInfo();
- times[pointNumInPage] = timeValuePairIterator.getTime();
- for (int i = 0; i < dataTypes.size(); i++) {
- pageColumnAccessInfo[i].add(accessInfo[i]);
+ TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
+ times[pointNumInPage] = tvPair.getTimestamp();
+ TsPrimitiveType[] values = tvPair.getValue().getVector();
+
+ for (int columnIndex = 0; columnIndex < values.length; columnIndex++) {
+ ValueChunkWriter valueChunkWriter =
+ alignedChunkWriter.getValueChunkWriterByIndex(columnIndex);
+ boolean isNull = values[columnIndex] == null;
+ switch (schemaList.get(columnIndex).getType()) {
+ case BOOLEAN:
+ valueChunkWriter.write(tvPair.getTimestamp(),
values[columnIndex].getBoolean(), isNull);
+ break;
+ case INT32:
+ case DATE:
+ valueChunkWriter.write(tvPair.getTimestamp(),
values[columnIndex].getInt(), isNull);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ valueChunkWriter.write(tvPair.getTimestamp(),
values[columnIndex].getLong(), isNull);
+ break;
+ case FLOAT:
+ valueChunkWriter.write(tvPair.getTimestamp(),
values[columnIndex].getFloat(), isNull);
+ break;
+ case DOUBLE:
+ valueChunkWriter.write(tvPair.getTimestamp(),
values[columnIndex].getDouble(), isNull);
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ valueChunkWriter.write(tvPair.getTimestamp(),
values[columnIndex].getBinary(), isNull);
+ break;
+ default:
+ break;
+ }
}
- timeValuePairIterator.step();
+
pointNumInPage++;
pointNumInChunk++;
if (pointNumInPage == MAX_NUMBER_OF_POINTS_IN_PAGE
|| pointNumInChunk >= maxNumberOfPointsInChunk) {
- writePageValuesIntoWriter(
- alignedChunkWriter, times, pageColumnAccessInfo,
timeValuePairIterator);
alignedChunkWriter.write(times, pointNumInPage, 0);
- for (PageColumnAccessInfo columnAccessInfo : pageColumnAccessInfo) {
- columnAccessInfo.reset();
- }
pointNumInPage = 0;
}
@@ -842,8 +782,6 @@ public class AlignedWritableMemChunk implements
IWritableMemChunk {
// last batch of points
if (pointNumInChunk > 0) {
if (pointNumInPage > 0) {
- writePageValuesIntoWriter(
- alignedChunkWriter, times, pageColumnAccessInfo,
timeValuePairIterator);
alignedChunkWriter.write(times, pointNumInPage, 0);
alignedChunkWriter.sealCurrentPage();
alignedChunkWriter.clearPageWriter();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
index 0e636619798..bc2031c069a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java
@@ -139,11 +139,6 @@ public class AlignedWritableMemChunkGroup implements
IWritableMemChunkGroup {
return deletedPointsNumber;
}
- @Override
- public long getMeasurementSize(String measurement) {
- return memChunk.rowCount();
- }
-
@Override
public IWritableMemChunk getWritableMemChunk(String measurement) {
return memChunk;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
index 27536102c26..9de36b73eae 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
@@ -173,8 +173,6 @@ public interface IMemTable extends WALEntryValue {
boolean checkIfChunkDoesNotExist(IDeviceID deviceId, String measurement);
/** only used when mem control enabled */
- long getMeasurementSize(IDeviceID deviceId, String measurement);
-
IWritableMemChunk getWritableMemChunk(IDeviceID deviceId, String
measurement);
/** only used when mem control enabled */
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
index 87cea400c39..79baafe86fd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IWritableMemChunkGroup.java
@@ -51,8 +51,6 @@ public interface IWritableMemChunkGroup extends WALEntryValue
{
int delete(
PartialPath originalPath, PartialPath devicePath, long startTimestamp,
long endTimestamp);
- long getMeasurementSize(String measurement);
-
IWritableMemChunk getWritableMemChunk(String measurement);
long getMaxTime();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
index 6247f5ea74e..a22f1da890a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
@@ -24,7 +24,7 @@ import
org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import
org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemChunkLoader;
import org.apache.iotdb.db.utils.MathUtils;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.tsfile.common.conf.TSFileDescriptor;
@@ -85,7 +85,7 @@ public class ReadOnlyMemChunk {
// TVList and its rowCount during query
private Map<TVList, Integer> tvListQueryMap;
- private MergeSortTvListIterator timeValuePairIterator;
+ private MergeSortTVListIterator timeValuePairIterator;
protected final int MAX_NUMBER_OF_POINTS_IN_PAGE =
TSFileDescriptor.getInstance().getConfig().getMaxNumberOfPointsInPage();
@@ -145,13 +145,13 @@ public class ReadOnlyMemChunk {
public void initChunkMetaFromTvLists() {
// create chunk statistics
Statistics<? extends Serializable> chunkStatistics =
Statistics.getStatsByType(dataType);
- int cnt = 0;
+ int pointsInChunk = 0;
int[] deleteCursor = {0};
List<TVList> tvLists = new ArrayList<>(tvListQueryMap.keySet());
- timeValuePairIterator = new MergeSortTvListIterator(tvLists,
floatPrecision, encoding);
+ timeValuePairIterator = new MergeSortTVListIterator(tvLists,
floatPrecision, encoding);
int[] tvListOffsets = timeValuePairIterator.getTVListOffsets();
while (timeValuePairIterator.hasNextTimeValuePair()) {
- if (cnt % MAX_NUMBER_OF_POINTS_IN_PAGE == 0) {
+ if (pointsInChunk % MAX_NUMBER_OF_POINTS_IN_PAGE == 0) {
Statistics<? extends Serializable> stats =
Statistics.getStatsByType(dataType);
pageStatisticsList.add(stats);
pageOffsetsList.add(Arrays.copyOf(tvListOffsets,
tvListOffsets.length));
@@ -194,7 +194,7 @@ public class ReadOnlyMemChunk {
String.format("Data type %s is not supported.", dataType));
}
}
- cnt++;
+ pointsInChunk++;
}
pageOffsetsList.add(Arrays.copyOf(tvListOffsets, tvListOffsets.length));
@@ -245,7 +245,7 @@ public class ReadOnlyMemChunk {
int[] deleteCursor = {0};
List<TVList> tvLists = new ArrayList<>(tvListQueryMap.keySet());
IPointReader timeValuePairIterator =
- new MergeSortTvListIterator(tvLists, floatPrecision, encoding);
+ new MergeSortTVListIterator(tvLists, floatPrecision, encoding);
while (timeValuePairIterator.hasNextTimeValuePair()) {
TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
@@ -333,7 +333,7 @@ public class ReadOnlyMemChunk {
return null;
}
- public MergeSortTvListIterator getMergeSortTVListIterator() {
+ public MergeSortTVListIterator getMergeSortTVListIterator() {
return timeValuePairIterator;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index 57f3bb5212c..6e214f76fc5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -581,8 +581,8 @@ public class TsFileProcessor {
memTableIncrement += TVList.tvListArrayMemCost(dataTypes[i]);
} else {
// here currentChunkPointNum >= 1
- long currentChunkPointNum = workMemTable.getMeasurementSize(deviceId,
measurements[i]);
IWritableMemChunk memChunk =
workMemTable.getWritableMemChunk(deviceId, measurements[i]);
+ long currentChunkPointNum = memChunk != null ? memChunk.rowCount() : 0;
if (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE == 0) {
memTableIncrement +=
memChunk != null ?
memChunk.getWorkingTVList().tvListArrayMemCost() : 0;
@@ -627,12 +627,12 @@ public class TsFileProcessor {
.putIfAbsent(measurements[i], 1);
} else {
// here currentChunkPointNum >= 1
- long currentChunkPointNum =
workMemTable.getMeasurementSize(deviceId, measurements[i]);
IWritableMemChunk memChunk =
workMemTable.getWritableMemChunk(deviceId, measurements[i]);
int addingPointNum =
increasingMemTableInfo
.computeIfAbsent(deviceId, k -> new HashMap<>())
.computeIfAbsent(measurements[i], k -> 0);
+ long currentChunkPointNum = memChunk != null ? memChunk.rowCount() :
0;
if ((currentChunkPointNum + addingPointNum) %
PrimitiveArrayManager.ARRAY_SIZE == 0) {
memTableIncrement +=
memChunk != null
@@ -873,8 +873,8 @@ public class TsFileProcessor {
((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
* TVList.tvListArrayMemCost(dataType);
} else {
- long currentChunkPointNum = workMemTable.getMeasurementSize(deviceId,
measurement);
IWritableMemChunk memChunk = workMemTable.getWritableMemChunk(deviceId,
measurement);
+ long currentChunkPointNum = memChunk != null ? memChunk.rowCount() : 0;
if (currentChunkPointNum % PrimitiveArrayManager.ARRAY_SIZE == 0) {
memIncrements[0] +=
((end - start) / PrimitiveArrayManager.ARRAY_SIZE + 1)
@@ -886,11 +886,7 @@ public class TsFileProcessor {
(end - start - 1 + (currentChunkPointNum %
PrimitiveArrayManager.ARRAY_SIZE))
/ PrimitiveArrayManager.ARRAY_SIZE;
if (acquireArray != 0) {
- memIncrements[0] +=
- acquireArray
- * (memChunk != null
- ? memChunk.getWorkingTVList().tvListArrayMemCost()
- : TVList.tvListArrayMemCost(dataType));
+ memIncrements[0] += acquireArray *
memChunk.getWorkingTVList().tvListArrayMemCost();
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
index ca5171f93e1..bc26cf72b00 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunk.java
@@ -25,7 +25,7 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import
org.apache.iotdb.db.queryengine.plan.planner.memory.MemoryReservationManager;
import
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
import org.apache.iotdb.db.utils.ModificationUtils;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.tsfile.enums.TSDataType;
@@ -119,34 +119,32 @@ public class WritableMemChunk implements
IWritableMemChunk {
}
@Override
- public synchronized void writeNonAlignedPoint(long insertTime, Object
objectValue) {
- synchronized (list) {
- switch (schema.getType()) {
- case BOOLEAN:
- putBoolean(insertTime, (boolean) objectValue);
- break;
- case INT32:
- case DATE:
- putInt(insertTime, (int) objectValue);
- break;
- case INT64:
- case TIMESTAMP:
- putLong(insertTime, (long) objectValue);
- break;
- case FLOAT:
- putFloat(insertTime, (float) objectValue);
- break;
- case DOUBLE:
- putDouble(insertTime, (double) objectValue);
- break;
- case TEXT:
- case BLOB:
- case STRING:
- putBinary(insertTime, (Binary) objectValue);
- break;
- default:
- throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
schema.getType().name());
- }
+ public void writeNonAlignedPoint(long insertTime, Object objectValue) {
+ switch (schema.getType()) {
+ case BOOLEAN:
+ putBoolean(insertTime, (boolean) objectValue);
+ break;
+ case INT32:
+ case DATE:
+ putInt(insertTime, (int) objectValue);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ putLong(insertTime, (long) objectValue);
+ break;
+ case FLOAT:
+ putFloat(insertTime, (float) objectValue);
+ break;
+ case DOUBLE:
+ putDouble(insertTime, (double) objectValue);
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ putBinary(insertTime, (Binary) objectValue);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
schema.getType().name());
}
if (TVLIST_SORT_THRESHOLD > 0 && list.rowCount() >= TVLIST_SORT_THRESHOLD)
{
handoverTvList();
@@ -160,41 +158,39 @@ public class WritableMemChunk implements
IWritableMemChunk {
}
@Override
- public synchronized void writeNonAlignedTablet(
+ public void writeNonAlignedTablet(
long[] times, Object valueList, BitMap bitMap, TSDataType dataType, int
start, int end) {
- synchronized (list) {
- switch (dataType) {
- case BOOLEAN:
- boolean[] boolValues = (boolean[]) valueList;
- putBooleans(times, boolValues, bitMap, start, end);
- break;
- case INT32:
- case DATE:
- int[] intValues = (int[]) valueList;
- putInts(times, intValues, bitMap, start, end);
- break;
- case INT64:
- case TIMESTAMP:
- long[] longValues = (long[]) valueList;
- putLongs(times, longValues, bitMap, start, end);
- break;
- case FLOAT:
- float[] floatValues = (float[]) valueList;
- putFloats(times, floatValues, bitMap, start, end);
- break;
- case DOUBLE:
- double[] doubleValues = (double[]) valueList;
- putDoubles(times, doubleValues, bitMap, start, end);
- break;
- case TEXT:
- case BLOB:
- case STRING:
- Binary[] binaryValues = (Binary[]) valueList;
- putBinaries(times, binaryValues, bitMap, start, end);
- break;
- default:
- throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
dataType.name());
- }
+ switch (dataType) {
+ case BOOLEAN:
+ boolean[] boolValues = (boolean[]) valueList;
+ putBooleans(times, boolValues, bitMap, start, end);
+ break;
+ case INT32:
+ case DATE:
+ int[] intValues = (int[]) valueList;
+ putInts(times, intValues, bitMap, start, end);
+ break;
+ case INT64:
+ case TIMESTAMP:
+ long[] longValues = (long[]) valueList;
+ putLongs(times, longValues, bitMap, start, end);
+ break;
+ case FLOAT:
+ float[] floatValues = (float[]) valueList;
+ putFloats(times, floatValues, bitMap, start, end);
+ break;
+ case DOUBLE:
+ double[] doubleValues = (double[]) valueList;
+ putDoubles(times, doubleValues, bitMap, start, end);
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ Binary[] binaryValues = (Binary[]) valueList;
+ putBinaries(times, binaryValues, bitMap, start, end);
+ break;
+ default:
+ throw new UnSupportedDataTypeException(UNSUPPORTED_TYPE +
dataType.name());
}
if (TVLIST_SORT_THRESHOLD > 0 && list.rowCount() >= TVLIST_SORT_THRESHOLD)
{
handoverTvList();
@@ -636,7 +632,7 @@ public class WritableMemChunk implements IWritableMemChunk {
// create MergeSortTvListIterator. It need not handle float/double
precision here.
List<TVList> tvLists = new ArrayList<>(sortedList);
tvLists.add(list);
- MergeSortTvListIterator timeValuePairIterator = new
MergeSortTvListIterator(tvLists);
+ MergeSortTVListIterator timeValuePairIterator = new
MergeSortTVListIterator(tvLists);
TimeValuePair prevTvPair = null;
while (timeValuePairIterator.hasNextTimeValuePair()) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
index 403098308df..313fcd71540 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java
@@ -153,14 +153,6 @@ public class WritableMemChunkGroup implements
IWritableMemChunkGroup {
return deletedPointsNumber;
}
- @Override
- public long getMeasurementSize(String measurement) {
- if (!memChunkMap.containsKey(measurement)) {
- return 0;
- }
- return memChunkMap.get(measurement).rowCount();
- }
-
@Override
public IWritableMemChunk getWritableMemChunk(String measurement) {
if (!memChunkMap.containsKey(measurement)) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
index 7d0c7eeea9a..ae27c462df0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
@@ -21,11 +21,11 @@ package
org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk;
import
org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedReadOnlyMemChunk;
import org.apache.iotdb.db.utils.datastructure.MergeSortAlignedTVListIterator;
-import org.apache.iotdb.db.utils.datastructure.PageColumnAccessInfo;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.statistics.Statistics;
+import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.common.BatchData;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.common.block.TsBlock;
@@ -142,115 +142,11 @@ public class MemAlignedChunkReader implements
IChunkReader {
return true;
}
- private void writePageTimeIntoBuilder(long[] time, int count,
TsBlockBuilder builder) {
- for (int index = 0; index < count; index++) {
- builder.getTimeColumnBuilder().writeLong(time[index]);
- }
- }
-
- private void writePageValuesIntoBuilder(
- PageColumnAccessInfo[] columnAccessInfo,
- List<TSDataType> tsDataTypes,
- TsBlockBuilder builder) {
- for (int columnIndex = 0; columnIndex < tsDataTypes.size();
columnIndex++) {
- PageColumnAccessInfo pageAccessInfo = columnAccessInfo[columnIndex];
- ColumnBuilder valueBuilder = builder.getColumnBuilder(columnIndex);
- switch (tsDataTypes.get(columnIndex)) {
- case BOOLEAN:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value == null) {
- valueBuilder.appendNull();
- } else {
- valueBuilder.writeBoolean(value.getBoolean());
- }
- }
- break;
- case INT32:
- case DATE:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value == null) {
- valueBuilder.appendNull();
- } else {
- valueBuilder.writeInt(value.getInt());
- }
- }
- break;
- case INT64:
- case TIMESTAMP:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value == null) {
- valueBuilder.appendNull();
- } else {
- valueBuilder.writeLong(value.getLong());
- }
- }
- break;
- case FLOAT:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value == null) {
- valueBuilder.appendNull();
- } else {
- valueBuilder.writeFloat(value.getFloat());
- }
- }
- break;
- case DOUBLE:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value == null) {
- valueBuilder.appendNull();
- } else {
- valueBuilder.writeDouble(value.getDouble());
- }
- }
- break;
- case TEXT:
- case BLOB:
- case STRING:
- for (int index = 0; index < pageAccessInfo.count(); index++) {
- int[] accessInfo = pageAccessInfo.get(index);
- TsPrimitiveType value =
- timeValuePairIterator.getPrimitiveObject(accessInfo,
columnIndex);
- if (value == null) {
- valueBuilder.appendNull();
- } else {
- valueBuilder.writeBinary(value.getBinary());
- }
- }
- break;
- default:
- break;
- }
- }
- }
-
// read one page and write to tsblock
private synchronized void writeValidValuesIntoTsBlock(TsBlockBuilder
builder) {
List<TSDataType> tsDataTypes = readableChunk.getDataTypes();
List<List<TimeRange>> valueColumnsDeletionList =
readableChunk.getValueColumnsDeletionList();
- int pointsInPage = 0;
- long[] time = new long[readableChunk.getMaxNumberOfPointsInPage()];
- PageColumnAccessInfo[] pageColumnAccessInfo = new
PageColumnAccessInfo[tsDataTypes.size()];
- for (int i = 0; i < pageColumnAccessInfo.length; i++) {
- pageColumnAccessInfo[i] =
- new
PageColumnAccessInfo(readableChunk.getMaxNumberOfPointsInPage());
- }
-
int[] timeDeleteCursor = new int[] {0};
List<int[]> valueColumnDeleteCursor = new ArrayList<>();
if (valueColumnsDeletionList != null) {
@@ -262,18 +158,20 @@ public class MemAlignedChunkReader implements
IChunkReader {
break;
}
- // skip deleted rows
- long timestamp = timeValuePairIterator.getTime();
-
- BitMap bitMap = timeValuePairIterator.getBitmap();
- if (valueColumnsDeletionList != null) {
- for (int columnIndex = 0; columnIndex < tsDataTypes.size();
columnIndex++) {
- if (isPointDeleted(
- timestamp,
- valueColumnsDeletionList.get(columnIndex),
- valueColumnDeleteCursor.get(columnIndex))) {
- bitMap.mark(columnIndex);
- }
+ TimeValuePair tvPair = timeValuePairIterator.nextTimeValuePair();
+ TsPrimitiveType[] values = tvPair.getValue().getVector();
+
+ BitMap bitMap = new BitMap(tsDataTypes.size());
+ for (int columnIndex = 0; columnIndex < tsDataTypes.size();
columnIndex++) {
+ if (values[columnIndex] == null) {
+ bitMap.mark(columnIndex);
+ } else if (valueColumnsDeletionList != null
+ && isPointDeleted(
+ tvPair.getTimestamp(),
+ valueColumnsDeletionList.get(columnIndex),
+ valueColumnDeleteCursor.get(columnIndex))) {
+ values[columnIndex] = null;
+ bitMap.mark(columnIndex);
}
}
if (bitMap.isAllMarked()) {
@@ -281,20 +179,50 @@ public class MemAlignedChunkReader implements
IChunkReader {
continue;
}
- // prepare column access info for current page
- int[][] accessInfo = timeValuePairIterator.getColumnAccessInfo();
- time[pointsInPage] = timeValuePairIterator.getTime();
- for (int i = 0; i < tsDataTypes.size(); i++) {
- pageColumnAccessInfo[i].add(accessInfo[i]);
+ // time column
+ builder.getTimeColumnBuilder().writeLong(tvPair.getTimestamp());
+ // value columns
+ for (int columnIndex = 0; columnIndex < values.length; columnIndex++) {
+ if (values[columnIndex] == null) {
+ builder.getColumnBuilder(columnIndex).appendNull();
+ continue;
+ }
+ ColumnBuilder valueBuilder = builder.getColumnBuilder(columnIndex);
+ switch (tsDataTypes.get(columnIndex)) {
+ case BOOLEAN:
+ valueBuilder.writeBoolean(values[columnIndex].getBoolean());
+ break;
+ case INT32:
+ case DATE:
+ valueBuilder.writeInt(values[columnIndex].getInt());
+ break;
+ case INT64:
+ case TIMESTAMP:
+ valueBuilder.writeLong(values[columnIndex].getLong());
+ break;
+ case FLOAT:
+ valueBuilder.writeFloat(values[columnIndex].getFloat());
+ break;
+ case DOUBLE:
+ valueBuilder.writeDouble(values[columnIndex].getDouble());
+ break;
+ case TEXT:
+ case BLOB:
+ case STRING:
+ valueBuilder.writeBinary(values[columnIndex].getBinary());
+ break;
+ default:
+ break;
+ }
}
- timeValuePairIterator.step();
- pointsInPage++;
+ builder.declarePosition();
+ }
+ if (builder.getPositionCount() >
readableChunk.getMaxNumberOfPointsInPage()) {
+ throw new RuntimeException(
+ String.format(
+ "Points in current page %d is larger than %d",
+ builder.getPositionCount(),
readableChunk.getMaxNumberOfPointsInPage()));
}
-
- // write time and values into builders
- writePageTimeIntoBuilder(time, pointsInPage, builder);
- writePageValuesIntoBuilder(pageColumnAccessInfo, tsDataTypes, builder);
- builder.declarePositions(pointsInPage);
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
index 96516415055..6bc1c3ba5c4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk;
import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IChunkMetadata;
@@ -47,7 +47,7 @@ import static
org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted;
public class MemChunkReader implements IChunkReader, IPointReader {
private final ReadOnlyMemChunk readableChunk;
- private final MergeSortTvListIterator timeValuePairIterator;
+ private final MergeSortTVListIterator timeValuePairIterator;
private final Filter globalTimeFilter;
private final List<IPageReader> pageReaderList;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
index ac46d6c5ff4..26e95ecbda6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
@@ -20,7 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk;
import
org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.metadata.PageMetadata;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
@@ -51,7 +51,7 @@ public class MemPageReader implements IPageReader {
private TsBlock tsBlock;
private Filter recordFilter;
- private final MergeSortTvListIterator mergeSortTvListIterator;
+ private final MergeSortTVListIterator mergeSortTvListIterator;
// MemPage range - [pageStartOffsets, pageEndOffsets)
private final int[] pageStartOffsets;
private final int[] pageEndOffsets;
@@ -63,7 +63,7 @@ public class MemPageReader implements IPageReader {
public MemPageReader(
Supplier<TsBlock> tsBlockSupplier,
- MergeSortTvListIterator mergeSortTvListIterator,
+ MergeSortTVListIterator mergeSortTvListIterator,
int[] pageStartOffsets,
int[] pageEndOffSets,
TSDataType tsDataType,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
index 4d80c40d718..c03cbd9f770 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
@@ -45,6 +45,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
+import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static
org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
@@ -1322,10 +1323,10 @@ public abstract class AlignedTVList extends TVList {
public class AlignedTVListIterator extends TVListIterator {
private BitMap allValueColDeletedMap;
- private TSDataType[] dataTypeArray;
- private int[] columnIndexArray;
+ private List<TSDataType> dataTypeList;
+ private List<Integer> columnIndexList;
private Integer floatPrecision;
- private TSEncoding[] encodingArray;
+ private List<TSEncoding> encodingList;
// remember the selected index of last not-null value for each column
during prepareNext phase.
// It is already converted by getValueIndex method, so it can be directly
used in
@@ -1342,14 +1343,14 @@ public abstract class AlignedTVList extends TVList {
Integer floatPrecision,
List<TSEncoding> encodingList) {
super(null, null);
- this.dataTypeArray = dataTypeList.toArray(new TSDataType[0]);
- this.columnIndexArray =
+ this.dataTypeList = dataTypeList;
+ this.columnIndexList =
(columnIndexList == null)
- ? IntStream.range(0, dataTypes.size()).toArray()
- : columnIndexList.stream().mapToInt(Integer::intValue).toArray();
+ ? IntStream.range(0,
dataTypes.size()).boxed().collect(Collectors.toList())
+ : columnIndexList;
this.allValueColDeletedMap = getAllValueColDeletedMap();
this.floatPrecision = floatPrecision;
- this.encodingArray = encodingList == null ? null :
encodingList.toArray(new TSEncoding[0]);
+ this.encodingList = encodingList;
this.selectedIndex = new int[dataTypeList.size()];
}
@@ -1380,7 +1381,7 @@ public abstract class AlignedTVList extends TVList {
// skip all-Null rows if allValueColDeletedMap exists
int rowIndex = getValueIndex(index);
if (allValueColDeletedMap == null ||
!allValueColDeletedMap.isMarked(rowIndex)) {
- for (int columnIndex = 0; columnIndex < dataTypeArray.length;
columnIndex++) {
+ for (int columnIndex = 0; columnIndex < dataTypeList.size();
columnIndex++) {
// update currTvPair if the column is not null
if (!isNull(rowIndex, columnIndex)) {
selectedIndex[columnIndex] = rowIndex;
@@ -1410,8 +1411,8 @@ public abstract class AlignedTVList extends TVList {
return null;
}
- TsPrimitiveType[] vector = new TsPrimitiveType[dataTypeArray.length];
- for (int columnIndex = 0; columnIndex < dataTypeArray.length;
columnIndex++) {
+ TsPrimitiveType[] vector = new TsPrimitiveType[dataTypeList.size()];
+ for (int columnIndex = 0; columnIndex < dataTypeList.size();
columnIndex++) {
vector[columnIndex] = getPrimitiveObject(selectedIndex[columnIndex],
columnIndex);
}
TimeValuePair tvPair =
@@ -1426,15 +1427,15 @@ public abstract class AlignedTVList extends TVList {
if (!hasCurrent()) {
return null;
}
- TsPrimitiveType[] vector = new TsPrimitiveType[dataTypeArray.length];
- for (int columnIndex = 0; columnIndex < dataTypeArray.length;
columnIndex++) {
+ TsPrimitiveType[] vector = new TsPrimitiveType[dataTypeList.size()];
+ for (int columnIndex = 0; columnIndex < dataTypeList.size();
columnIndex++) {
vector[columnIndex] = getPrimitiveObject(selectedIndex[columnIndex],
columnIndex);
}
return new TimeValuePair(currentTime,
TsPrimitiveType.getByType(TSDataType.VECTOR, vector));
}
public boolean isNull(int rowIndex, int columnIndex) {
- int validColumnIndex = columnIndexArray[columnIndex];
+ int validColumnIndex = columnIndexList.get(columnIndex);
if (validColumnIndex < 0 || validColumnIndex >= dataTypes.size()) {
return true;
}
@@ -1445,14 +1446,14 @@ public abstract class AlignedTVList extends TVList {
if (rowIndex < 0 || rowIndex >= rows) {
return null;
}
- int validColumnIndex = columnIndexArray[columnIndex];
+ int validColumnIndex = columnIndexList.get(columnIndex);
if (validColumnIndex < 0 || validColumnIndex >= dataTypes.size()) {
return null;
}
if (isNullValue(rowIndex, validColumnIndex)) {
return null;
}
- switch (dataTypeArray[columnIndex]) {
+ switch (dataTypeList.get(columnIndex)) {
case BOOLEAN:
return TsPrimitiveType.getByType(
TSDataType.BOOLEAN, getBooleanByValueIndex(rowIndex,
validColumnIndex));
@@ -1467,20 +1468,20 @@ public abstract class AlignedTVList extends TVList {
case FLOAT:
float valueF = getFloatByValueIndex(rowIndex, validColumnIndex);
if (floatPrecision != null
- && encodingArray != null
+ && encodingList != null
&& !Float.isNaN(valueF)
- && (encodingArray[columnIndex] == TSEncoding.RLE
- || encodingArray[columnIndex] == TSEncoding.TS_2DIFF)) {
+ && (encodingList.get(columnIndex) == TSEncoding.RLE
+ || encodingList.get(columnIndex) == TSEncoding.TS_2DIFF)) {
valueF = MathUtils.roundWithGivenPrecision(valueF, floatPrecision);
}
return TsPrimitiveType.getByType(TSDataType.FLOAT, valueF);
case DOUBLE:
double valueD = getDoubleByValueIndex(rowIndex, validColumnIndex);
if (floatPrecision != null
- && encodingArray != null
+ && encodingList != null
&& !Double.isNaN(valueD)
- && (encodingArray[columnIndex] == TSEncoding.RLE
- || encodingArray[columnIndex] == TSEncoding.TS_2DIFF)) {
+ && (encodingList.get(columnIndex) == TSEncoding.RLE
+ || encodingList.get(columnIndex) == TSEncoding.TS_2DIFF)) {
valueD = MathUtils.roundWithGivenPrecision(valueD, floatPrecision);
}
return TsPrimitiveType.getByType(TSDataType.DOUBLE, valueD);
@@ -1491,7 +1492,7 @@ public abstract class AlignedTVList extends TVList {
TSDataType.TEXT, getBinaryByValueIndex(rowIndex,
validColumnIndex));
default:
throw new UnSupportedDataTypeException(
- String.format("Data type %s is not supported.",
dataTypeArray[columnIndex]));
+ String.format("Data type %s is not supported.",
dataTypeList.get(columnIndex)));
}
}
@@ -1503,12 +1504,12 @@ public abstract class AlignedTVList extends TVList {
public AlignedTVListIterator clone() {
AlignedTVListIterator iterator = new AlignedTVListIterator();
iterator.rows = rows;
- iterator.dataTypeArray = dataTypeArray;
- iterator.columnIndexArray = columnIndexArray;
+ iterator.dataTypeList = dataTypeList;
+ iterator.columnIndexList = columnIndexList;
iterator.allValueColDeletedMap = allValueColDeletedMap;
iterator.floatPrecision = floatPrecision;
- iterator.encodingArray = this.encodingArray;
- iterator.selectedIndex = new int[dataTypeArray.length];
+ iterator.encodingList = this.encodingList;
+ iterator.selectedIndex = new int[dataTypeList.size()];
iterator.reset();
return iterator;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
index 1c5a1265a84..fd26eded1ac 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BinaryTVList.java
@@ -246,7 +246,6 @@ public abstract class BinaryTVList extends TVList {
tIdx = tIdx - nullCnt;
inPutMinTime = Math.min(inPutMinTime, time[tIdx]);
maxTime = Math.max(maxTime, time[tIdx]);
- minTime = Math.min(minTime, time[tIdx]);
if (inputSorted) {
if (tIdx > 0 && time[tIdx - 1] > time[tIdx]) {
inputSorted = false;
@@ -255,6 +254,7 @@ public abstract class BinaryTVList extends TVList {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted
&& (rowCount == 0
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
index 0037bf72308..2a84c13f546 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BooleanTVList.java
@@ -246,7 +246,6 @@ public abstract class BooleanTVList extends TVList {
tIdx = tIdx - nullCnt;
inPutMinTime = Math.min(inPutMinTime, time[tIdx]);
maxTime = Math.max(maxTime, time[tIdx]);
- minTime = Math.min(minTime, time[tIdx]);
if (inputSorted) {
if (tIdx > 0 && time[tIdx - 1] > time[tIdx]) {
inputSorted = false;
@@ -255,6 +254,7 @@ public abstract class BooleanTVList extends TVList {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted
&& (rowCount == 0
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
index 48c7c58046d..5f82527ca76 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/DoubleTVList.java
@@ -252,7 +252,6 @@ public abstract class DoubleTVList extends TVList {
tIdx = tIdx - nullCnt;
inPutMinTime = Math.min(inPutMinTime, time[tIdx]);
maxTime = Math.max(maxTime, time[tIdx]);
- minTime = Math.min(minTime, time[tIdx]);
if (inputSorted) {
if (tIdx > 0 && time[tIdx - 1] > time[tIdx]) {
inputSorted = false;
@@ -261,6 +260,7 @@ public abstract class DoubleTVList extends TVList {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted
&& (rowCount == 0
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
index a0a23bd2b1c..de79955a2e7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FloatTVList.java
@@ -252,7 +252,6 @@ public abstract class FloatTVList extends TVList {
tIdx = tIdx - nullCnt;
inPutMinTime = Math.min(inPutMinTime, time[tIdx]);
maxTime = Math.max(maxTime, time[tIdx]);
- minTime = Math.min(minTime, time[tIdx]);
if (inputSorted) {
if (tIdx > 0 && time[tIdx - 1] > time[tIdx]) {
inputSorted = false;
@@ -261,6 +260,7 @@ public abstract class FloatTVList extends TVList {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted
&& (rowCount == 0
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
index 5611ea700e8..f3312ebd9c2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/IntTVList.java
@@ -244,7 +244,6 @@ public abstract class IntTVList extends TVList {
tIdx = tIdx - nullCnt;
inPutMinTime = Math.min(inPutMinTime, time[tIdx]);
maxTime = Math.max(maxTime, time[tIdx]);
- minTime = Math.min(minTime, time[tIdx]);
if (inputSorted) {
if (tIdx > 0 && time[tIdx - 1] > time[tIdx]) {
inputSorted = false;
@@ -253,6 +252,7 @@ public abstract class IntTVList extends TVList {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted
&& (rowCount == 0
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
index e5f360c6635..bea59266fcd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LongTVList.java
@@ -244,7 +244,6 @@ public abstract class LongTVList extends TVList {
tIdx = tIdx - nullCnt;
inPutMinTime = Math.min(inPutMinTime, time[tIdx]);
maxTime = Math.max(maxTime, time[tIdx]);
- minTime = Math.min(minTime, time[tIdx]);
if (inputSorted) {
if (tIdx > 0 && time[tIdx - 1] > time[tIdx]) {
inputSorted = false;
@@ -253,6 +252,7 @@ public abstract class LongTVList extends TVList {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted
&& (rowCount == 0
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortAlignedTVListIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortAlignedTVListIterator.java
index fc418c6930d..98f444abeff 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortAlignedTVListIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortAlignedTVListIterator.java
@@ -23,37 +23,30 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.reader.IPointReader;
-import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.utils.TsPrimitiveType;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class MergeSortAlignedTVListIterator implements IPointReader {
- private final AlignedTVList.AlignedTVListIterator[] alignedTvListIterators;
- private final int columnNum;
+ private List<AlignedTVList.AlignedTVListIterator> alignedTvListIterators;
private boolean probeNext = false;
- private boolean hasNext = false;
- private final List<Integer> probeIterators;
+ private TimeValuePair currentTvPair;
+
+ private List<Integer> probeIterators;
// Min-Heap: minimal timestamp; if same timestamp, maximum TVList index
private final PriorityQueue<Pair<Long, Integer>> minHeap =
new PriorityQueue<>(
(a, b) -> a.left.equals(b.left) ? b.right.compareTo(a.right) :
a.left.compareTo(b.left));
- private final int[] alignedTvListOffsets;
-
- // Remember the selected columns for prepareNext
- // column index -> [selectedTVList, selectedIndex]
- private final int[][] columnAccessInfo;
-
- private long time;
- private final BitMap bitMap;
+ private int[] alignedTvListOffsets;
public MergeSortAlignedTVListIterator(
List<AlignedTVList> alignedTvLists,
@@ -61,45 +54,31 @@ public class MergeSortAlignedTVListIterator implements
IPointReader {
List<Integer> columnIndexList,
Integer floatPrecision,
List<TSEncoding> encodingList) {
- this.alignedTvListIterators = new
AlignedTVList.AlignedTVListIterator[alignedTvLists.size()];
- for (int i = 0; i < alignedTvLists.size(); i++) {
- alignedTvListIterators[i] =
- alignedTvLists
- .get(i)
- .iterator(tsDataTypes, columnIndexList, floatPrecision,
encodingList);
+ this.alignedTvListIterators = new ArrayList<>(alignedTvLists.size());
+ for (AlignedTVList alignedTvList : alignedTvLists) {
+ alignedTvListIterators.add(
+ alignedTvList.iterator(tsDataTypes, columnIndexList, floatPrecision,
encodingList));
}
this.alignedTvListOffsets = new int[alignedTvLists.size()];
- this.columnNum = tsDataTypes.size();
- this.columnAccessInfo = new int[columnNum][];
- for (int i = 0; i < columnAccessInfo.length; i++) {
- columnAccessInfo[i] = new int[2];
- }
- this.bitMap = new BitMap(columnNum);
this.probeIterators =
- IntStream.range(0,
alignedTvListIterators.length).boxed().collect(Collectors.toList());
+ IntStream.range(0,
alignedTvListIterators.size()).boxed().collect(Collectors.toList());
}
- public MergeSortAlignedTVListIterator(
- AlignedTVList.AlignedTVListIterator[] alignedTvListIterators, int
columnNum) {
- this.alignedTvListIterators =
- new AlignedTVList.AlignedTVListIterator[alignedTvListIterators.length];
- for (int i = 0; i < alignedTvListIterators.length; i++) {
- this.alignedTvListIterators[i] = alignedTvListIterators[i].clone();
- }
- this.alignedTvListOffsets = new int[alignedTvListIterators.length];
- this.columnNum = columnNum;
- this.columnAccessInfo = new int[columnNum][];
- for (int i = 0; i < columnAccessInfo.length; i++) {
- columnAccessInfo[i] = new int[2];
- }
- this.bitMap = new BitMap(columnNum);
- this.probeIterators =
- IntStream.range(0,
alignedTvListIterators.length).boxed().collect(Collectors.toList());
- }
+ private MergeSortAlignedTVListIterator() {}
private void prepareNextRow() {
+ currentTvPair = null;
+ if (alignedTvListIterators.size() == 1) {
+ AlignedTVList.AlignedTVListIterator iterator =
alignedTvListIterators.get(0);
+ if (iterator.hasNext()) {
+ currentTvPair = iterator.current();
+ }
+ probeNext = true;
+ return;
+ }
+
for (int i : probeIterators) {
- TVList.TVListIterator iterator = alignedTvListIterators[i];
+ AlignedTVList.AlignedTVListIterator iterator =
alignedTvListIterators.get(i);
if (iterator.hasNext()) {
minHeap.add(new Pair<>(iterator.currentTime(), i));
}
@@ -108,32 +87,24 @@ public class MergeSortAlignedTVListIterator implements
IPointReader {
if (!minHeap.isEmpty()) {
Pair<Long, Integer> top = minHeap.poll();
- time = top.left;
+ long time = top.left;
probeIterators.add(top.right);
- for (int columnIndex = 0; columnIndex < columnNum; columnIndex++) {
- int rowIndex =
alignedTvListIterators[top.right].getSelectedIndex(columnIndex);
- columnAccessInfo[columnIndex][0] = top.right;
- columnAccessInfo[columnIndex][1] = rowIndex;
- if (alignedTvListIterators[top.right].isNull(rowIndex, columnIndex)) {
- bitMap.mark(columnIndex);
- }
- }
+ currentTvPair = alignedTvListIterators.get(top.right).current();
+ TsPrimitiveType[] currentValues = currentTvPair.getValue().getVector();
+
while (!minHeap.isEmpty() && minHeap.peek().left == time) {
Pair<Long, Integer> element = minHeap.poll();
probeIterators.add(element.right);
- for (int columnIndex = 0; columnIndex < columnNum; columnIndex++) {
+
+ TimeValuePair tvPair =
alignedTvListIterators.get(element.right).current();
+ TsPrimitiveType[] values = tvPair.getValue().getVector();
+ for (int columnIndex = 0; columnIndex < values.length; columnIndex++) {
// if the column is currently not null, it needs not update
- if (bitMap.isMarked(columnIndex)) {
- int rowIndex =
alignedTvListIterators[element.right].getSelectedIndex(columnIndex);
- if (!alignedTvListIterators[element.right].isNull(rowIndex,
columnIndex)) {
- columnAccessInfo[columnIndex][0] = element.right;
- columnAccessInfo[columnIndex][1] = rowIndex;
- bitMap.unmark(columnIndex);
- }
+ if (currentValues[columnIndex] == null) {
+ currentValues[columnIndex] = values[columnIndex];
}
}
}
- hasNext = true;
}
probeNext = true;
}
@@ -143,7 +114,7 @@ public class MergeSortAlignedTVListIterator implements
IPointReader {
if (!probeNext) {
prepareNextRow();
}
- return hasNext;
+ return currentTvPair != null;
}
@Override
@@ -151,9 +122,8 @@ public class MergeSortAlignedTVListIterator implements
IPointReader {
if (!hasNextTimeValuePair()) {
return null;
}
- TimeValuePair tvPair = buildTimeValuePair();
step();
- return tvPair;
+ return currentTvPair;
}
@Override
@@ -161,28 +131,22 @@ public class MergeSortAlignedTVListIterator implements
IPointReader {
if (!hasNextTimeValuePair()) {
return null;
}
- return buildTimeValuePair();
- }
-
- private TimeValuePair buildTimeValuePair() {
- TsPrimitiveType[] vector = new TsPrimitiveType[columnNum];
- for (int columnIndex = 0; columnIndex < vector.length; columnIndex++) {
- int[] accessInfo = columnAccessInfo[columnIndex];
- AlignedTVList.AlignedTVListIterator iterator =
alignedTvListIterators[accessInfo[0]];
- vector[columnIndex] = iterator.getPrimitiveObject(accessInfo[1],
columnIndex);
- }
- return new TimeValuePair(time,
TsPrimitiveType.getByType(TSDataType.VECTOR, vector));
+ return currentTvPair;
}
public void step() {
- for (int index : probeIterators) {
- TVList.TVListIterator iterator = alignedTvListIterators[index];
+ if (alignedTvListIterators.size() == 1) {
+ AlignedTVList.AlignedTVListIterator iterator =
alignedTvListIterators.get(0);
iterator.step();
- alignedTvListOffsets[index] = iterator.getIndex();
+ alignedTvListOffsets[0] = iterator.getIndex();
+ } else {
+ for (int index : probeIterators) {
+ AlignedTVList.AlignedTVListIterator iterator =
alignedTvListIterators.get(index);
+ iterator.step();
+ alignedTvListOffsets[index] = iterator.getIndex();
+ }
}
probeNext = false;
- hasNext = false;
- bitMap.reset();
}
@Override
@@ -199,42 +163,30 @@ public class MergeSortAlignedTVListIterator implements
IPointReader {
}
public void setAlignedTVListOffsets(int[] alignedTvListOffsets) {
- for (int i = 0; i < alignedTvListIterators.length; i++) {
- alignedTvListIterators[i].setIndex(alignedTvListOffsets[i]);
+ for (int i = 0; i < alignedTvListIterators.size(); i++) {
+ alignedTvListIterators.get(i).setIndex(alignedTvListOffsets[i]);
this.alignedTvListOffsets[i] = alignedTvListOffsets[i];
}
- minHeap.clear();
- probeIterators.clear();
- for (int i = 0; i < alignedTvListIterators.length; i++) {
- probeIterators.add(i);
+ if (alignedTvListIterators.size() > 1) {
+ minHeap.clear();
+ probeIterators.clear();
+ for (int i = 0; i < alignedTvListIterators.size(); i++) {
+ probeIterators.add(i);
+ }
}
probeNext = false;
- hasNext = false;
- bitMap.reset();
- }
-
- public int[][] getColumnAccessInfo() {
- return columnAccessInfo;
- }
-
- public long getTime() {
- return time;
- }
-
- public TsPrimitiveType getPrimitiveObject(int[] accessInfo, int columnIndex)
{
- if (columnIndex >= columnAccessInfo.length) {
- return null;
- }
- AlignedTVList.AlignedTVListIterator iterator =
alignedTvListIterators[accessInfo[0]];
- return iterator.getPrimitiveObject(accessInfo[1], columnIndex);
- }
-
- public BitMap getBitmap() {
- return bitMap;
}
@Override
public MergeSortAlignedTVListIterator clone() {
- return new MergeSortAlignedTVListIterator(alignedTvListIterators,
columnNum);
+ MergeSortAlignedTVListIterator cloneIterator = new
MergeSortAlignedTVListIterator();
+ cloneIterator.alignedTvListIterators = new
ArrayList<>(alignedTvListIterators.size());
+ for (int i = 0; i < alignedTvListIterators.size(); i++) {
+
cloneIterator.alignedTvListIterators.add(alignedTvListIterators.get(i).clone());
+ }
+ cloneIterator.alignedTvListOffsets = new
int[alignedTvListIterators.size()];
+ cloneIterator.probeIterators =
+ IntStream.range(0,
alignedTvListIterators.size()).boxed().collect(Collectors.toList());
+ return cloneIterator;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTvListIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTVListIterator.java
similarity index 58%
rename from
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTvListIterator.java
rename to
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTVListIterator.java
index 342752a7088..7c4834efe9c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTvListIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortTVListIterator.java
@@ -25,58 +25,60 @@ import org.apache.tsfile.read.reader.IPointReader;
import org.apache.tsfile.utils.Pair;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-public class MergeSortTvListIterator implements IPointReader {
- private final TVList.TVListIterator[] tvListIterators;
- private final int[] tvListOffsets;
+public class MergeSortTVListIterator implements IPointReader {
+ private List<TVList.TVListIterator> tvListIterators;
+ private int[] tvListOffsets;
private boolean probeNext = false;
private TimeValuePair currentTvPair;
- private final List<Integer> probeIterators;
+ private List<Integer> probeIterators;
private final PriorityQueue<Pair<Long, Integer>> minHeap =
new PriorityQueue<>(
(a, b) -> a.left.equals(b.left) ? b.right.compareTo(a.right) :
a.left.compareTo(b.left));
- public MergeSortTvListIterator(List<TVList> tvLists) {
- tvListIterators = new TVList.TVListIterator[tvLists.size()];
- for (int i = 0; i < tvLists.size(); i++) {
- tvListIterators[i] = tvLists.get(i).iterator(null, null);
+ public MergeSortTVListIterator(List<TVList> tvLists) {
+ tvListIterators = new ArrayList<>(tvLists.size());
+ for (TVList tvList : tvLists) {
+ tvListIterators.add(tvList.iterator(null, null));
}
this.tvListOffsets = new int[tvLists.size()];
this.probeIterators =
- IntStream.range(0,
tvListIterators.length).boxed().collect(Collectors.toList());
+ IntStream.range(0,
tvListIterators.size()).boxed().collect(Collectors.toList());
}
- public MergeSortTvListIterator(
+ public MergeSortTVListIterator(
List<TVList> tvLists, Integer floatPrecision, TSEncoding encoding) {
- tvListIterators = new TVList.TVListIterator[tvLists.size()];
- for (int i = 0; i < tvLists.size(); i++) {
- tvListIterators[i] = tvLists.get(i).iterator(floatPrecision, encoding);
+ tvListIterators = new ArrayList<>(tvLists.size());
+ for (TVList tvList : tvLists) {
+ tvListIterators.add(tvList.iterator(floatPrecision, encoding));
}
this.tvListOffsets = new int[tvLists.size()];
this.probeIterators =
- IntStream.range(0,
tvListIterators.length).boxed().collect(Collectors.toList());
+ IntStream.range(0,
tvListIterators.size()).boxed().collect(Collectors.toList());
}
- public MergeSortTvListIterator(TVList.TVListIterator[] tvListIterators) {
- this.tvListIterators = new TVList.TVListIterator[tvListIterators.length];
- for (int i = 0; i < tvListIterators.length; i++) {
- this.tvListIterators[i] = tvListIterators[i].clone();
- }
- this.tvListOffsets = new int[tvListIterators.length];
- this.probeIterators =
- IntStream.range(0,
tvListIterators.length).boxed().collect(Collectors.toList());
- }
+ private MergeSortTVListIterator() {}
private void prepareNext() {
currentTvPair = null;
+ if (tvListIterators.size() == 1) {
+ TVList.TVListIterator iterator = tvListIterators.get(0);
+ if (iterator.hasNext()) {
+ currentTvPair = iterator.current();
+ }
+ probeNext = true;
+ return;
+ }
+
for (int i : probeIterators) {
- TVList.TVListIterator iterator = tvListIterators[i];
+ TVList.TVListIterator iterator = tvListIterators.get(i);
if (iterator.hasNext()) {
minHeap.add(new Pair<>(iterator.currentTime(), i));
}
@@ -86,7 +88,7 @@ public class MergeSortTvListIterator implements IPointReader {
if (!minHeap.isEmpty()) {
Pair<Long, Integer> top = minHeap.poll();
probeIterators.add(top.right);
- currentTvPair = tvListIterators[top.right].current();
+ currentTvPair = tvListIterators.get(top.right).current();
while (!minHeap.isEmpty() && minHeap.peek().left.longValue() ==
top.left.longValue()) {
Pair<Long, Integer> element = minHeap.poll();
probeIterators.add(element.right);
@@ -121,10 +123,16 @@ public class MergeSortTvListIterator implements
IPointReader {
}
public void step() {
- for (int index : probeIterators) {
- TVList.TVListIterator iterator = tvListIterators[index];
+ if (tvListIterators.size() == 1) {
+ TVList.TVListIterator iterator = tvListIterators.get(0);
iterator.step();
- tvListOffsets[index] = iterator.getIndex();
+ tvListOffsets[0] = iterator.getIndex();
+ } else {
+ for (int index : probeIterators) {
+ TVList.TVListIterator iterator = tvListIterators.get(index);
+ iterator.step();
+ tvListOffsets[index] = iterator.getIndex();
+ }
}
probeNext = false;
}
@@ -143,20 +151,30 @@ public class MergeSortTvListIterator implements
IPointReader {
}
public void setTVListOffsets(int[] tvListOffsets) {
- for (int i = 0; i < tvListIterators.length; i++) {
- tvListIterators[i].setIndex(tvListOffsets[i]);
+ for (int i = 0; i < tvListIterators.size(); i++) {
+ tvListIterators.get(i).setIndex(tvListOffsets[i]);
this.tvListOffsets[i] = tvListOffsets[i];
}
- minHeap.clear();
- probeIterators.clear();
- for (int i = 0; i < tvListIterators.length; i++) {
- probeIterators.add(i);
+ if (tvListIterators.size() > 1) {
+ minHeap.clear();
+ probeIterators.clear();
+ for (int i = 0; i < tvListIterators.size(); i++) {
+ probeIterators.add(i);
+ }
}
probeNext = false;
}
@Override
- public MergeSortTvListIterator clone() {
- return new MergeSortTvListIterator(tvListIterators);
+ public MergeSortTVListIterator clone() {
+ MergeSortTVListIterator cloneIterator = new MergeSortTVListIterator();
+ cloneIterator.tvListIterators = new ArrayList<>(tvListIterators.size());
+ for (int i = 0; i < tvListIterators.size(); i++) {
+ cloneIterator.tvListIterators.add(tvListIterators.get(i).clone());
+ }
+ cloneIterator.tvListOffsets = new int[tvListIterators.size()];
+ cloneIterator.probeIterators =
+ IntStream.range(0,
tvListIterators.size()).boxed().collect(Collectors.toList());
+ return cloneIterator;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/PageColumnAccessInfo.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/PageColumnAccessInfo.java
deleted file mode 100644
index 39d8976932b..00000000000
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/PageColumnAccessInfo.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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;
-
-public class PageColumnAccessInfo {
- // time -> (selectedTVList, selectedIndex)
- private final int[][] indices;
- private int count;
-
- public PageColumnAccessInfo(int maxNumberOfPointsInPage) {
- this.indices = new int[maxNumberOfPointsInPage][];
- for (int i = 0; i < maxNumberOfPointsInPage; i++) {
- indices[i] = new int[2];
- }
- this.count = 0;
- }
-
- public int[] get(int index) {
- return indices[index];
- }
-
- public void add(int[] columnAccess) {
- indices[count][0] = columnAccess[0];
- indices[count][1] = columnAccess[1];
- count++;
- }
-
- public int count() {
- return count;
- }
-
- public void reset() {
- count = 0;
- }
-}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
index c4714fa4620..bdbcaefd238 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -413,7 +413,7 @@ public abstract class TVList implements WALEntryValue {
}
// common clone for both TVList and AlignedTVList
- protected void cloneAs(TVList cloneList) {
+ protected synchronized void cloneAs(TVList cloneList) {
// clone timestamps
for (long[] timestampArray : timestamps) {
cloneList.timestamps.add(cloneTime(timestampArray));
@@ -497,7 +497,6 @@ public abstract class TVList implements WALEntryValue {
for (int i = start; i < end; i++) {
inPutMinTime = Math.min(inPutMinTime, time[i]);
maxTime = Math.max(maxTime, time[i]);
- minTime = Math.min(minTime, time[i]);
if (inputSorted) {
if (i < length - 1 && time[i] > time[i + 1]) {
inputSorted = false;
@@ -506,6 +505,7 @@ public abstract class TVList implements WALEntryValue {
}
}
}
+ minTime = Math.min(minTime, inPutMinTime);
if (sorted && (rowCount == 0 || time[start] >= getTime(rowCount - 1))) {
seqRowCount += inputSeqRowCount;
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkLoaderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkLoaderTest.java
index 8462909c49e..b56f2720ef0 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkLoaderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkLoaderTest.java
@@ -21,7 +21,7 @@ package
org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
-import org.apache.iotdb.db.utils.datastructure.MergeSortTvListIterator;
+import org.apache.iotdb.db.utils.datastructure.MergeSortTVListIterator;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.tsfile.common.conf.TSFileConfig;
@@ -68,8 +68,8 @@ public class MemChunkLoaderTest {
Map<TVList, Integer> booleanTvListMap = buildBooleanTvListMap();
Mockito.when(chunk.getTvListQueryMap()).thenReturn(booleanTvListMap);
List<TVList> booleanTvLists = new ArrayList<>(booleanTvListMap.keySet());
- MergeSortTvListIterator timeValuePairIterator =
- new MergeSortTvListIterator(booleanTvLists, null, null);
+ MergeSortTVListIterator timeValuePairIterator =
+ new MergeSortTVListIterator(booleanTvLists, null, null);
Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -148,8 +148,8 @@ public class MemChunkLoaderTest {
Map<TVList, Integer> int32TvListMap = buildInt32TvListMap();
Mockito.when(chunk.getTvListQueryMap()).thenReturn(int32TvListMap);
List<TVList> int32TvLists = new ArrayList<>(int32TvListMap.keySet());
- MergeSortTvListIterator timeValuePairIterator =
- new MergeSortTvListIterator(int32TvLists, null, null);
+ MergeSortTVListIterator timeValuePairIterator =
+ new MergeSortTVListIterator(int32TvLists, null, null);
Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -229,8 +229,8 @@ public class MemChunkLoaderTest {
Map<TVList, Integer> int64TvListMap = buildInt64TvListMap();
Mockito.when(chunk.getTvListQueryMap()).thenReturn(int64TvListMap);
List<TVList> int64TvLists = new ArrayList<>(int64TvListMap.keySet());
- MergeSortTvListIterator timeValuePairIterator =
- new MergeSortTvListIterator(int64TvLists, null, null);
+ MergeSortTVListIterator timeValuePairIterator =
+ new MergeSortTVListIterator(int64TvLists, null, null);
Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -310,8 +310,8 @@ public class MemChunkLoaderTest {
Map<TVList, Integer> floatTvListMap = buildFloatTvListMap();
Mockito.when(chunk.getTvListQueryMap()).thenReturn(floatTvListMap);
List<TVList> floatTvLists = new ArrayList<>(floatTvListMap.keySet());
- MergeSortTvListIterator timeValuePairIterator =
- new MergeSortTvListIterator(floatTvLists, null, null);
+ MergeSortTVListIterator timeValuePairIterator =
+ new MergeSortTVListIterator(floatTvLists, null, null);
Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -391,8 +391,8 @@ public class MemChunkLoaderTest {
Map<TVList, Integer> doubleTvListMap = buildDoubleTvListMap();
Mockito.when(chunk.getTvListQueryMap()).thenReturn(doubleTvListMap);
List<TVList> doubleTvLists = new ArrayList<>(doubleTvListMap.keySet());
- MergeSortTvListIterator timeValuePairIterator =
- new MergeSortTvListIterator(doubleTvLists, null, null);
+ MergeSortTVListIterator timeValuePairIterator =
+ new MergeSortTVListIterator(doubleTvLists, null, null);
Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
@@ -472,8 +472,8 @@ public class MemChunkLoaderTest {
Map<TVList, Integer> textTvListMap = buildTextTvListMap();
Mockito.when(chunk.getTvListQueryMap()).thenReturn(textTvListMap);
List<TVList> textTvLists = new ArrayList<>(textTvListMap.keySet());
- MergeSortTvListIterator timeValuePairIterator =
- new MergeSortTvListIterator(textTvLists, null, null);
+ MergeSortTVListIterator timeValuePairIterator =
+ new MergeSortTVListIterator(textTvLists, null, null);
Mockito.when(chunk.getMergeSortTVListIterator()).thenReturn(timeValuePairIterator);
ChunkMetadata chunkMetadata = Mockito.mock(ChunkMetadata.class);
