This is an automated email from the ASF dual-hosted git repository.
Caideyipi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a51258bbb97 [Performance] Reduce aligned flush duplicate timestamp
overhead
a51258bbb97 is described below
commit a51258bbb979b0601e1688fb8f88adc65d518c93
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 28 15:44:39 2026 +0800
[Performance] Reduce aligned flush duplicate timestamp overhead
---
.../memtable/AlignedWritableMemChunk.java | 119 ++++++-----
...gnedFlushDuplicateTimeCheckPerformanceTest.java | 238 +++++++++++++++++++++
.../dataregion/memtable/MemTableFlushTaskTest.java | 83 +++++++
3 files changed, 386 insertions(+), 54 deletions(-)
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 e96baf88b3a..ebdc024aac8 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
@@ -709,41 +709,41 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
TypeServices.StorageEngine.ALIGNED_TV_LIST_CHUNK_WRITER_SERVICE.call(
Type.fromTsDataType(dataType)));
}
- Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[dataTypes.size()];
+ long[] lastValidPointTimeForTimeDupCheck =
+ timeDuplicateInfo == null ? null : new long[dataTypes.size()];
+ int[] lastValidPointIndexForTimeDupCheck =
+ timeDuplicateInfo == null ? null : new int[dataTypes.size()];
+ if (lastValidPointIndexForTimeDupCheck != null) {
+ Arrays.fill(lastValidPointIndexForTimeDupCheck, -1);
+ }
for (List<Integer> pageRange : chunkRange) {
AlignedChunkWriterImpl alignedChunkWriter =
new AlignedChunkWriterImpl(schemaList, encryptParameter);
for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
for (int columnIndex = 0; columnIndex < dataTypes.size();
columnIndex++) {
- // Pair of Time and Index
- if (Objects.nonNull(timeDuplicateInfo)
- && lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
- lastValidPointIndexForTimeDupCheck[columnIndex] = new
Pair<>(Long.MIN_VALUE, null);
- }
TypeServices.AlignedTVListChunkWriter valueWriter =
valueWriters.get(columnIndex);
ValueChunkWriter valueChunkWriter =
alignedChunkWriter.getValueChunkWriterByIndex(columnIndex);
for (int sortedRowIndex = pageRange.get(pageNum * 2);
sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
sortedRowIndex++) {
+ int valueIndex =
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
// skip empty row
- if (allValueColDeletedMap != null
- && allValueColDeletedMap.isMarked(
- alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))
{
+ if (allValueColDeletedMap != null &&
allValueColDeletedMap.isMarked(valueIndex)) {
continue;
}
// Keep value pages aligned with the time page when an entire
timestamp is deleted.
- if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex)) {
+ if (alignedWorkingListForFlush.isTimeDeleted(valueIndex, false)) {
continue;
}
// skip time duplicated rows
long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
- if (Objects.nonNull(timeDuplicateInfo)) {
- if (!alignedWorkingListForFlush.isNullValue(
- alignedWorkingListForFlush.getValueIndex(sortedRowIndex),
columnIndex)) {
- lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
- lastValidPointIndexForTimeDupCheck[columnIndex].right =
- alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ boolean currentValueIsNull = false;
+ if (timeDuplicateInfo != null) {
+ currentValueIsNull =
alignedWorkingListForFlush.isNullValue(valueIndex, columnIndex);
+ if (!currentValueIsNull) {
+ lastValidPointTimeForTimeDupCheck[columnIndex] = time;
+ lastValidPointIndexForTimeDupCheck[columnIndex] = valueIndex;
}
if (timeDuplicateInfo[sortedRowIndex]) {
continue;
@@ -753,21 +753,25 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
// The part of code solves the following problem:
// Time: 1,2,2,3
// Value: 1,2,null,null
- // When rowIndex:1, pair(min,null), timeDuplicateInfo:false,
write(T:1,V:1)
- // When rowIndex:2, pair(2,2), timeDuplicateInfo:true, skip
writing value
- // When rowIndex:3, pair(2,2), timeDuplicateInfo:false,
T:2==pair.left:2, write(T:2,V:2)
- // When rowIndex:4, pair(2,2), timeDuplicateInfo:false,
T:3!=pair.left:2,
+ // When rowIndex:1, no valid duplicate value,
timeDuplicateInfo:false, write(T:1,V:1)
+ // When rowIndex:2, last valid point=(2,2),
timeDuplicateInfo:true, skip writing value
+ // When rowIndex:3, last valid point=(2,2), T:2 matches,
write(T:2,V:2)
+ // When rowIndex:4, last valid point=(2,2), T:3 doesn't match,
// write(T:3,V:null)
int originRowIndex;
- if
(Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
- && (time ==
lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
- originRowIndex =
lastValidPointIndexForTimeDupCheck[columnIndex].right;
+ if (timeDuplicateInfo != null
+ && lastValidPointIndexForTimeDupCheck[columnIndex] >= 0
+ && time == lastValidPointTimeForTimeDupCheck[columnIndex]) {
+ originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex];
} else {
- originRowIndex =
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ originRowIndex = valueIndex;
}
- boolean isNull =
alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex);
+ boolean isNull =
+ timeDuplicateInfo == null
+ ? alignedWorkingListForFlush.isNullValue(originRowIndex,
columnIndex)
+ : originRowIndex == valueIndex && currentValueIsNull;
valueWriter.write(
valueChunkWriter,
time,
@@ -947,18 +951,19 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
List<IMeasurementSchema> activeSchemaList) {
AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
- Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[activeSchemaList.size()];
+ long[] lastValidPointTimeForTimeDupCheck =
+ timeDuplicateInfo == null ? null : new long[activeSchemaList.size()];
+ int[] lastValidPointIndexForTimeDupCheck =
+ timeDuplicateInfo == null ? null : new int[activeSchemaList.size()];
+ if (lastValidPointIndexForTimeDupCheck != null) {
+ Arrays.fill(lastValidPointIndexForTimeDupCheck, -1);
+ }
for (List<Integer> pageRange : chunkRange) {
AlignedChunkWriterImpl alignedChunkWriter =
new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
for (int columnIndex = 0; columnIndex < activeSchemaList.size();
columnIndex++) {
int tvListColumnIndex = columnIndexList.get(columnIndex);
- // Pair of Time and Index
- if (Objects.nonNull(timeDuplicateInfo)
- && lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
- lastValidPointIndexForTimeDupCheck[columnIndex] = new
Pair<>(Long.MIN_VALUE, null);
- }
TSDataType tsDataType = activeSchemaList.get(columnIndex).getType();
TypeServices.AlignedTVListColumnWriter columnWriter =
TypeServices.StorageEngine.ALIGNED_TV_LIST_COLUMN_WRITER_SERVICE.call(
@@ -966,27 +971,27 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
for (int sortedRowIndex = pageRange.get(pageNum * 2);
sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
sortedRowIndex++) {
+ int valueIndex =
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
// skip empty row
- if (allValueColDeletedMap != null
- && allValueColDeletedMap.isMarked(
- alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))
{
+ if (allValueColDeletedMap != null &&
allValueColDeletedMap.isMarked(valueIndex)) {
continue;
}
// Keep value pages aligned with the time page when an entire
timestamp is deleted.
- if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex)) {
+ if (alignedWorkingListForFlush.isTimeDeleted(valueIndex, false)) {
continue;
}
// skip time duplicated rows
long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
- if (Objects.nonNull(timeDuplicateInfo)) {
- if (tvListColumnIndex >= 0
- && !alignedWorkingListForFlush.isNullValue(
- alignedWorkingListForFlush.getValueIndex(sortedRowIndex),
- tvListColumnIndex)) {
- lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
- lastValidPointIndexForTimeDupCheck[columnIndex].right =
- alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ boolean currentValueIsNull = true;
+ if (timeDuplicateInfo != null && tvListColumnIndex >= 0) {
+ currentValueIsNull =
+ alignedWorkingListForFlush.isNullValue(valueIndex,
tvListColumnIndex);
+ if (!currentValueIsNull) {
+ lastValidPointTimeForTimeDupCheck[columnIndex] = time;
+ lastValidPointIndexForTimeDupCheck[columnIndex] = valueIndex;
}
+ }
+ if (timeDuplicateInfo != null) {
if (timeDuplicateInfo[sortedRowIndex]) {
continue;
}
@@ -995,24 +1000,30 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
// The part of code solves the following problem:
// Time: 1,2,2,3
// Value: 1,2,null,null
- // When rowIndex:1, pair(min,null), timeDuplicateInfo:false,
write(T:1,V:1)
- // When rowIndex:2, pair(2,2), timeDuplicateInfo:true, skip
writing value
- // When rowIndex:3, pair(2,2), timeDuplicateInfo:false,
T:2==pair.left:2, write(T:2,V:2)
- // When rowIndex:4, pair(2,2), timeDuplicateInfo:false,
T:3!=pair.left:2,
+ // When rowIndex:1, no valid duplicate value,
timeDuplicateInfo:false, write(T:1,V:1)
+ // When rowIndex:2, last valid point=(2,2),
timeDuplicateInfo:true, skip writing value
+ // When rowIndex:3, last valid point=(2,2), T:2 matches,
write(T:2,V:2)
+ // When rowIndex:4, last valid point=(2,2), T:3 doesn't match,
// write(T:3,V:null)
int originRowIndex;
if (tvListColumnIndex >= 0
- &&
Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
- && (time ==
lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
- originRowIndex =
lastValidPointIndexForTimeDupCheck[columnIndex].right;
+ && timeDuplicateInfo != null
+ && lastValidPointIndexForTimeDupCheck[columnIndex] >= 0
+ && time == lastValidPointTimeForTimeDupCheck[columnIndex]) {
+ originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex];
} else {
- originRowIndex =
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ originRowIndex = valueIndex;
}
- boolean isNull =
- tvListColumnIndex < 0
- || alignedWorkingListForFlush.isNullValue(originRowIndex,
tvListColumnIndex);
+ boolean isNull;
+ if (tvListColumnIndex < 0) {
+ isNull = true;
+ } else if (timeDuplicateInfo == null) {
+ isNull = alignedWorkingListForFlush.isNullValue(originRowIndex,
tvListColumnIndex);
+ } else {
+ isNull = originRowIndex == valueIndex && currentValueIsNull;
+ }
columnWriter.write(
alignedChunkWriter,
time,
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedFlushDuplicateTimeCheckPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedFlushDuplicateTimeCheckPerformanceTest.java
new file mode 100644
index 00000000000..1f396efb6ff
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedFlushDuplicateTimeCheckPerformanceTest.java
@@ -0,0 +1,238 @@
+/*
+ * 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.storageengine.dataregion.memtable;
+
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary;
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Pair;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+
+public class AlignedFlushDuplicateTimeCheckPerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
"iotdb.aligned.flush.duplicate-check.perf.enabled";
+ private static final String ROUNDS_PROPERTY =
"iotdb.aligned.flush.duplicate-check.perf.rounds";
+ private static final String SCANS_PROPERTY =
"iotdb.aligned.flush.duplicate-check.perf.scans";
+ private static final int ROW_COUNT = 16_384;
+ private static final int COLUMN_COUNT = 8;
+ private static final int WARMUP_ROUNDS = 3;
+
+ private static volatile long benchmarkBlackhole;
+
+ @Test
+ public void duplicateTimeCheckBenchmark() {
+ Assume.assumeTrue(
+ String.format(
+ Locale.ROOT,
+ "Manual performance UT. Enable with -D%s=true; optionally tune
-D%s and -D%s.",
+ ENABLED_PROPERTY,
+ ROUNDS_PROPERTY,
+ SCANS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+ Assume.assumeTrue(
+ "Current-thread CPU time and allocation metrics are required.",
+ ManualPerformanceTestUtils.enableThreadMetrics());
+
+ int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5);
+ int scans = Integer.getInteger(SCANS_PROPERTY, 128);
+ Assert.assertTrue(rounds > 0);
+ Assert.assertTrue(scans > 0);
+
+ BenchmarkData data = createBenchmarkData();
+ long legacyResult = runLegacy(data);
+ long optimizedResult = runOptimized(data);
+ Assert.assertEquals(legacyResult, optimizedResult);
+
+ for (int i = 0; i < WARMUP_ROUNDS; i++) {
+ runLegacy(data);
+ runOptimized(data);
+ }
+
+ Measurement[] legacyMeasurements = new Measurement[rounds];
+ Measurement[] optimizedMeasurements = new Measurement[rounds];
+ for (int i = 0; i < rounds; i++) {
+ if ((i & 1) == 0) {
+ legacyMeasurements[i] = measureLegacy(data, scans);
+ optimizedMeasurements[i] = measureOptimized(data, scans);
+ } else {
+ optimizedMeasurements[i] = measureOptimized(data, scans);
+ legacyMeasurements[i] = measureLegacy(data, scans);
+ }
+ }
+
+ Summary legacySummary =
ManualPerformanceTestUtils.summarize(legacyMeasurements, scans);
+ Summary optimizedSummary =
ManualPerformanceTestUtils.summarize(optimizedMeasurements, scans);
+ printResult(rounds, scans, legacySummary, optimizedSummary);
+ }
+
+ private static Measurement measureLegacy(BenchmarkData data, int scans) {
+ return ManualPerformanceTestUtils.measure(scans, () -> runLegacy(data));
+ }
+
+ private static Measurement measureOptimized(BenchmarkData data, int scans) {
+ return ManualPerformanceTestUtils.measure(scans, () -> runOptimized(data));
+ }
+
+ private static long runLegacy(BenchmarkData data) {
+ long nullCount = 0;
+ Pair<Long, Integer>[] lastValidPointForTimeDupCheck = new
Pair[COLUMN_COUNT];
+ for (int columnIndex = 0; columnIndex < COLUMN_COUNT; columnIndex++) {
+ for (int rowIndex = 0; rowIndex < ROW_COUNT; rowIndex++) {
+ if (lastValidPointForTimeDupCheck[columnIndex] == null) {
+ lastValidPointForTimeDupCheck[columnIndex] = new
Pair<>(Long.MIN_VALUE, null);
+ }
+ int valueIndex = data.list.getValueIndex(rowIndex);
+ long time = data.list.getTime(rowIndex);
+ if (!data.list.isNullValue(valueIndex, columnIndex)) {
+ lastValidPointForTimeDupCheck[columnIndex].left = time;
+ lastValidPointForTimeDupCheck[columnIndex].right = valueIndex;
+ }
+ if (data.timeDuplicateInfo[rowIndex]) {
+ continue;
+ }
+
+ int originRowIndex;
+ if (time == lastValidPointForTimeDupCheck[columnIndex].left) {
+ originRowIndex = lastValidPointForTimeDupCheck[columnIndex].right;
+ } else {
+ originRowIndex = valueIndex;
+ }
+ if (data.list.isNullValue(originRowIndex, columnIndex)) {
+ nullCount++;
+ }
+ }
+ }
+ benchmarkBlackhole = nullCount;
+ return nullCount;
+ }
+
+ private static long runOptimized(BenchmarkData data) {
+ long nullCount = 0;
+ long[] lastValidPointTimeForTimeDupCheck = new long[COLUMN_COUNT];
+ int[] lastValidPointIndexForTimeDupCheck = new int[COLUMN_COUNT];
+ Arrays.fill(lastValidPointIndexForTimeDupCheck, -1);
+ for (int columnIndex = 0; columnIndex < COLUMN_COUNT; columnIndex++) {
+ for (int rowIndex = 0; rowIndex < ROW_COUNT; rowIndex++) {
+ int valueIndex = data.list.getValueIndex(rowIndex);
+ long time = data.list.getTime(rowIndex);
+ boolean currentValueIsNull = data.list.isNullValue(valueIndex,
columnIndex);
+ if (!currentValueIsNull) {
+ lastValidPointTimeForTimeDupCheck[columnIndex] = time;
+ lastValidPointIndexForTimeDupCheck[columnIndex] = valueIndex;
+ }
+ if (data.timeDuplicateInfo[rowIndex]) {
+ continue;
+ }
+
+ int originRowIndex;
+ if (lastValidPointIndexForTimeDupCheck[columnIndex] >= 0
+ && time == lastValidPointTimeForTimeDupCheck[columnIndex]) {
+ originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex];
+ } else {
+ originRowIndex = valueIndex;
+ }
+ boolean isNull = originRowIndex == valueIndex && currentValueIsNull;
+ if (isNull) {
+ nullCount++;
+ }
+ }
+ }
+ benchmarkBlackhole = nullCount;
+ return nullCount;
+ }
+
+ private static BenchmarkData createBenchmarkData() {
+ List<TSDataType> dataTypes = new ArrayList<>(COLUMN_COUNT);
+ for (int columnIndex = 0; columnIndex < COLUMN_COUNT; columnIndex++) {
+ dataTypes.add(TSDataType.INT64);
+ }
+ AlignedTVList list = AlignedTVList.newAlignedList(dataTypes);
+ for (int rowIndex = 0; rowIndex < ROW_COUNT; rowIndex++) {
+ Object[] values = new Object[COLUMN_COUNT];
+ for (int columnIndex = 0; columnIndex < COLUMN_COUNT; columnIndex++) {
+ if ((rowIndex + columnIndex) % 4 != 0) {
+ values[columnIndex] = (long) rowIndex + columnIndex;
+ }
+ }
+ list.putAlignedValue(rowIndex / 4, values);
+ }
+ list.sort();
+
+ boolean[] timeDuplicateInfo = new boolean[ROW_COUNT];
+ for (int rowIndex = 0; rowIndex < ROW_COUNT - 1; rowIndex++) {
+ timeDuplicateInfo[rowIndex] = list.getTime(rowIndex) ==
list.getTime(rowIndex + 1);
+ }
+ return new BenchmarkData(list, timeDuplicateInfo);
+ }
+
+ private static void printResult(
+ int rounds, int scans, Summary legacySummary, Summary optimizedSummary) {
+ System.out.printf(
+ Locale.ROOT,
+ "Aligned flush duplicate-time check benchmark: rows=%d, columns=%d,
value checks/scan=%d, scans/round=%d, rounds=%d%n",
+ ROW_COUNT,
+ COLUMN_COUNT,
+ ROW_COUNT * COLUMN_COUNT,
+ scans,
+ rounds);
+ printSummary("legacy", legacySummary);
+ printSummary("optimized", optimizedSummary);
+ System.out.printf(
+ Locale.ROOT,
+ " optimized/legacy CPU ratio=%.2f%%, allocation ratio=%.2f%%%n",
+ percentage(
+ optimizedSummary.getCpuNanosPerOperation(),
legacySummary.getCpuNanosPerOperation()),
+ percentage(
+ optimizedSummary.getAllocatedBytesPerOperation(),
+ legacySummary.getAllocatedBytesPerOperation()));
+ }
+
+ private static void printSummary(String label, Summary summary) {
+ System.out.printf(
+ Locale.ROOT,
+ " %-10s CPU=%.3f ms/scan, allocated=%.2f bytes/scan%n",
+ label,
+ summary.getCpuNanosPerOperation() / 1_000_000.0,
+ summary.getAllocatedBytesPerOperation());
+ }
+
+ private static double percentage(double numerator, double denominator) {
+ return denominator == 0 ? 0 : numerator * 100.0 / denominator;
+ }
+
+ private static final class BenchmarkData {
+ private final AlignedTVList list;
+ private final boolean[] timeDuplicateInfo;
+
+ private BenchmarkData(AlignedTVList list, boolean[] timeDuplicateInfo) {
+ this.list = list;
+ this.timeDuplicateInfo = timeDuplicateInfo;
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
index ad1cbc219de..afe0384839b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java
@@ -212,6 +212,89 @@ public class MemTableFlushTaskTest {
assertFalse(ioTaskQueue.isEmpty());
}
+ @Test
+ public void testAlignedFlushMergesDuplicateTimestampsUsingLastNonNullValue()
throws Exception {
+ checkAlignedFlushMergesDuplicateTimestamps(false);
+ }
+
+ @Test
+ public void
testAlignedFlushMergesDuplicateTimestampsAfterMeasurementDeletion() throws
Exception {
+ checkAlignedFlushMergesDuplicateTimestamps(true);
+ }
+
+ private void checkAlignedFlushMergesDuplicateTimestamps(boolean
removeSecondMeasurement)
+ throws Exception {
+ List<IMeasurementSchema> schemas =
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN),
+ new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN));
+ AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas,
true);
+ String alignedFilePath =
+ TestConstant.OUTPUT_DATA_DIR.concat(
+ "duplicateAlignedRows-" + removeSecondMeasurement + ".tsfile");
+
+ try {
+ memChunk.putAlignedRow(1, new Object[] {10L, 100L});
+ memChunk.putAlignedRow(2, new Object[] {20L, null});
+ memChunk.putAlignedRow(2, new Object[] {null, 200L});
+ memChunk.putAlignedRow(2, new Object[] {null, null});
+ memChunk.putAlignedRow(3, new Object[] {30L, null});
+ if (removeSecondMeasurement) {
+ memChunk.removeColumn("s1");
+ }
+ memChunk.sortTvListForFlush();
+
+ BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>();
+ memChunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100);
+ try (TsFileIOWriter alignedWriter = new TsFileIOWriter(new
File(alignedFilePath))) {
+
alignedWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d"));
+ Object task;
+ while ((task = ioTaskQueue.poll()) != null) {
+ if (task instanceof IChunkWriter chunkWriter) {
+ chunkWriter.writeToFileWriter(alignedWriter);
+ }
+ }
+ alignedWriter.endChunkGroup();
+ alignedWriter.endFile();
+ }
+
+ try (TsFileSequenceReader sequenceReader = new
TsFileSequenceReader(alignedFilePath);
+ TsFileReader fileReader = new TsFileReader(sequenceReader)) {
+ List<Path> paths = new ArrayList<>();
+ paths.add(new Path("root.d", "s0", false));
+ if (!removeSecondMeasurement) {
+ paths.add(new Path("root.d", "s1", false));
+ }
+ QueryDataSet dataSet = fileReader.query(QueryExpression.create(paths,
null));
+
+ RowRecord row = dataSet.next();
+ assertEquals(1, row.getTimestamp());
+ assertEquals(10L, row.getFields().get(0).getLongV());
+ if (!removeSecondMeasurement) {
+ assertEquals(100L, row.getFields().get(1).getLongV());
+ }
+
+ row = dataSet.next();
+ assertEquals(2, row.getTimestamp());
+ assertEquals(20L, row.getFields().get(0).getLongV());
+ if (!removeSecondMeasurement) {
+ assertEquals(200L, row.getFields().get(1).getLongV());
+ }
+
+ row = dataSet.next();
+ assertEquals(3, row.getTimestamp());
+ assertEquals(30L, row.getFields().get(0).getLongV());
+ if (!removeSecondMeasurement) {
+ assertTrue(
+ row.getFields().get(1) == null ||
row.getFields().get(1).getDataType() == null);
+ }
+ assertFalse(dataSet.hasNext());
+ }
+ } finally {
+ memChunk.release();
+ }
+ }
+
@Test
public void
testAlignedFastPathKeepsPagesAndValuesAlignedAfterPartialSegmentSort()
throws IOException, InterruptedException {
