Skip to content

Commit 160cd51

Browse files
CaideyipiCaideyipi
committed
[Performance] Reduce aligned flush duplicate timestamp overhead
(cherry picked from commit a51258b)
1 parent e3f50df commit 160cd51

3 files changed

Lines changed: 394 additions & 52 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java‎

Lines changed: 63 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -587,34 +587,34 @@ private void handleEncodingWithoutDeletedMeasurements(
587587
int maxNumberOfPointsInPage) {
588588
AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush;
589589
List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
590-
Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new Pair[dataTypes.size()];
590+
long[] lastValidPointTimeForTimeDupCheck =
591+
timeDuplicateInfo == null ? null : new long[dataTypes.size()];
592+
int[] lastValidPointIndexForTimeDupCheck =
593+
timeDuplicateInfo == null ? null : new int[dataTypes.size()];
594+
if (lastValidPointIndexForTimeDupCheck != null) {
595+
Arrays.fill(lastValidPointIndexForTimeDupCheck, -1);
596+
}
591597
for (List<Integer> pageRange : chunkRange) {
592598
AlignedChunkWriterImpl alignedChunkWriter = new AlignedChunkWriterImpl(schemaList);
593599
for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
594600
for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) {
595-
// Pair of Time and Index
596-
if (Objects.nonNull(timeDuplicateInfo)
597-
&& lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
598-
lastValidPointIndexForTimeDupCheck[columnIndex] = new Pair<>(Long.MIN_VALUE, null);
599-
}
600601
TSDataType tsDataType = dataTypes.get(columnIndex);
601602
for (int sortedRowIndex = pageRange.get(pageNum * 2);
602603
sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
603604
sortedRowIndex++) {
605+
int valueIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
604606
// skip empty row
605-
if (allValueColDeletedMap != null
606-
&& allValueColDeletedMap.isMarked(
607-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex))) {
607+
if (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(valueIndex)) {
608608
continue;
609609
}
610610
// skip time duplicated rows
611611
long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
612-
if (Objects.nonNull(timeDuplicateInfo)) {
613-
if (!alignedWorkingListForFlush.isNullValue(
614-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex), columnIndex)) {
615-
lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
616-
lastValidPointIndexForTimeDupCheck[columnIndex].right =
617-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
612+
boolean currentValueIsNull = false;
613+
if (timeDuplicateInfo != null) {
614+
currentValueIsNull = alignedWorkingListForFlush.isNullValue(valueIndex, columnIndex);
615+
if (!currentValueIsNull) {
616+
lastValidPointTimeForTimeDupCheck[columnIndex] = time;
617+
lastValidPointIndexForTimeDupCheck[columnIndex] = valueIndex;
618618
}
619619
if (timeDuplicateInfo[sortedRowIndex]) {
620620
continue;
@@ -624,21 +624,25 @@ private void handleEncodingWithoutDeletedMeasurements(
624624
// The part of code solves the following problem:
625625
// Time: 1,2,2,3
626626
// Value: 1,2,null,null
627-
// When rowIndex:1, pair(min,null), timeDuplicateInfo:false, write(T:1,V:1)
628-
// When rowIndex:2, pair(2,2), timeDuplicateInfo:true, skip writing value
629-
// When rowIndex:3, pair(2,2), timeDuplicateInfo:false, T:2==pair.left:2, write(T:2,V:2)
630-
// When rowIndex:4, pair(2,2), timeDuplicateInfo:false, T:3!=pair.left:2,
627+
// When rowIndex:1, no valid duplicate value, timeDuplicateInfo:false, write(T:1,V:1)
628+
// When rowIndex:2, last valid point=(2,2), timeDuplicateInfo:true, skip writing value
629+
// When rowIndex:3, last valid point=(2,2), T:2 matches, write(T:2,V:2)
630+
// When rowIndex:4, last valid point=(2,2), T:3 doesn't match,
631631
// write(T:3,V:null)
632632

633633
int originRowIndex;
634-
if (Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
635-
&& (time == lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
636-
originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex].right;
634+
if (timeDuplicateInfo != null
635+
&& lastValidPointIndexForTimeDupCheck[columnIndex] >= 0
636+
&& time == lastValidPointTimeForTimeDupCheck[columnIndex]) {
637+
originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex];
637638
} else {
638-
originRowIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
639+
originRowIndex = valueIndex;
639640
}
640641

641-
boolean isNull = alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex);
642+
boolean isNull =
643+
timeDuplicateInfo == null
644+
? alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex)
645+
: originRowIndex == valueIndex && currentValueIsNull;
642646
switch (tsDataType) {
643647
case BOOLEAN:
644648
alignedChunkWriter.writeByColumn(
@@ -741,38 +745,39 @@ private void handleEncodingWithDeletedMeasurements(
741745
List<IMeasurementSchema> activeSchemaList) {
742746
AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush;
743747
List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
744-
Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new Pair[activeSchemaList.size()];
748+
long[] lastValidPointTimeForTimeDupCheck =
749+
timeDuplicateInfo == null ? null : new long[activeSchemaList.size()];
750+
int[] lastValidPointIndexForTimeDupCheck =
751+
timeDuplicateInfo == null ? null : new int[activeSchemaList.size()];
752+
if (lastValidPointIndexForTimeDupCheck != null) {
753+
Arrays.fill(lastValidPointIndexForTimeDupCheck, -1);
754+
}
745755
for (List<Integer> pageRange : chunkRange) {
746756
AlignedChunkWriterImpl alignedChunkWriter = new AlignedChunkWriterImpl(activeSchemaList);
747757
for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
748758
for (int columnIndex = 0; columnIndex < activeSchemaList.size(); columnIndex++) {
749759
int tvListColumnIndex = columnIndexList.get(columnIndex);
750-
// Pair of Time and Index
751-
if (Objects.nonNull(timeDuplicateInfo)
752-
&& lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
753-
lastValidPointIndexForTimeDupCheck[columnIndex] = new Pair<>(Long.MIN_VALUE, null);
754-
}
755760
TSDataType tsDataType = activeSchemaList.get(columnIndex).getType();
756761
for (int sortedRowIndex = pageRange.get(pageNum * 2);
757762
sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
758763
sortedRowIndex++) {
764+
int valueIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
759765
// skip empty row
760-
if (allValueColDeletedMap != null
761-
&& allValueColDeletedMap.isMarked(
762-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex))) {
766+
if (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(valueIndex)) {
763767
continue;
764768
}
765769
// skip time duplicated rows
766770
long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
767-
if (Objects.nonNull(timeDuplicateInfo)) {
768-
if (tvListColumnIndex >= 0
769-
&& !alignedWorkingListForFlush.isNullValue(
770-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex),
771-
tvListColumnIndex)) {
772-
lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
773-
lastValidPointIndexForTimeDupCheck[columnIndex].right =
774-
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
771+
boolean currentValueIsNull = true;
772+
if (timeDuplicateInfo != null && tvListColumnIndex >= 0) {
773+
currentValueIsNull =
774+
alignedWorkingListForFlush.isNullValue(valueIndex, tvListColumnIndex);
775+
if (!currentValueIsNull) {
776+
lastValidPointTimeForTimeDupCheck[columnIndex] = time;
777+
lastValidPointIndexForTimeDupCheck[columnIndex] = valueIndex;
775778
}
779+
}
780+
if (timeDuplicateInfo != null) {
776781
if (timeDuplicateInfo[sortedRowIndex]) {
777782
continue;
778783
}
@@ -781,24 +786,30 @@ private void handleEncodingWithDeletedMeasurements(
781786
// The part of code solves the following problem:
782787
// Time: 1,2,2,3
783788
// Value: 1,2,null,null
784-
// When rowIndex:1, pair(min,null), timeDuplicateInfo:false, write(T:1,V:1)
785-
// When rowIndex:2, pair(2,2), timeDuplicateInfo:true, skip writing value
786-
// When rowIndex:3, pair(2,2), timeDuplicateInfo:false, T:2==pair.left:2, write(T:2,V:2)
787-
// When rowIndex:4, pair(2,2), timeDuplicateInfo:false, T:3!=pair.left:2,
789+
// When rowIndex:1, no valid duplicate value, timeDuplicateInfo:false, write(T:1,V:1)
790+
// When rowIndex:2, last valid point=(2,2), timeDuplicateInfo:true, skip writing value
791+
// When rowIndex:3, last valid point=(2,2), T:2 matches, write(T:2,V:2)
792+
// When rowIndex:4, last valid point=(2,2), T:3 doesn't match,
788793
// write(T:3,V:null)
789794

790795
int originRowIndex;
791796
if (tvListColumnIndex >= 0
792-
&& Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
793-
&& (time == lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
794-
originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex].right;
797+
&& timeDuplicateInfo != null
798+
&& lastValidPointIndexForTimeDupCheck[columnIndex] >= 0
799+
&& time == lastValidPointTimeForTimeDupCheck[columnIndex]) {
800+
originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex];
795801
} else {
796-
originRowIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
802+
originRowIndex = valueIndex;
797803
}
798804

799-
boolean isNull =
800-
tvListColumnIndex < 0
801-
|| alignedWorkingListForFlush.isNullValue(originRowIndex, tvListColumnIndex);
805+
boolean isNull;
806+
if (tvListColumnIndex < 0) {
807+
isNull = true;
808+
} else if (timeDuplicateInfo == null) {
809+
isNull = alignedWorkingListForFlush.isNullValue(originRowIndex, tvListColumnIndex);
810+
} else {
811+
isNull = originRowIndex == valueIndex && currentValueIsNull;
812+
}
802813
switch (tsDataType) {
803814
case BOOLEAN:
804815
alignedChunkWriter.writeByColumn(

0 commit comments

Comments
 (0)