This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 47844ddc737 Optimize aligned memtable encoding for null values (#18610)
47844ddc737 is described below
commit 47844ddc7372dd884f383884532009b71587ae48
Author: Jiang Tian <[email protected]>
AuthorDate: Thu Sep 10 11:30:43 2026 +0800
Optimize aligned memtable encoding for null values (#18610)
* use new tsfile interfaces
* fix test
---
.../ConcatDeviceAndBindSchemaForHavingVisitor.java | 14 +-
.../memtable/AlignedWritableMemChunk.java | 245 +++++++++++++++++++++
.../db/utils/datastructure/AlignedTVList.java | 49 +++++
.../db/utils/datastructure/BackAlignedTVList.java | 1 +
.../db/utils/datastructure/QuickAlignedTVList.java | 1 +
.../db/utils/datastructure/TimAlignedTVList.java | 3 +-
.../dataregion/memtable/MemTableFlushTaskTest.java | 190 ++++++++++++++++
.../db/utils/datastructure/AlignedTVListTest.java | 32 +++
pom.xml | 2 +-
9 files changed, 534 insertions(+), 3 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForHavingVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForHavingVisitor.java
index 7f1a1b7133c..266b74270e1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForHavingVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/expression/visitor/cartesian/ConcatDeviceAndBindSchemaForHavingVisitor.java
@@ -28,6 +28,9 @@ import
org.apache.iotdb.db.queryengine.plan.expression.Expression;
import org.apache.iotdb.db.queryengine.plan.expression.leaf.TimeSeriesOperand;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.write.schema.MeasurementSchema;
import java.util.ArrayList;
import java.util.Collections;
@@ -46,8 +49,17 @@ public class ConcatDeviceAndBindSchemaForHavingVisitor
List<MeasurementPath> actualPaths =
context.getSchemaTree().searchMeasurementPaths(concatPath).left;
if (actualPaths.isEmpty()) {
+ // Preserve UNKNOWN for missing-measurement analysis. This schema is
never written, so use
+ // explicit placeholders instead of asking TsFile for UNKNOWN's
unsupported defaults.
return Collections.singletonList(
- new TimeSeriesOperand(new MeasurementPath(concatPath,
TSDataType.UNKNOWN)));
+ new TimeSeriesOperand(
+ new MeasurementPath(
+ concatPath,
+ new MeasurementSchema(
+ concatPath.getMeasurement(),
+ TSDataType.UNKNOWN,
+ TSEncoding.PLAIN,
+ CompressionType.UNCOMPRESSED))));
}
List<MeasurementPath> nonViewActualPaths = new ArrayList<>();
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 752e7c92b63..d1c15384173 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
@@ -44,6 +44,7 @@ import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.UnSupportedDataTypeException;
import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl;
import org.apache.tsfile.write.chunk.IChunkWriter;
+import org.apache.tsfile.write.chunk.ValueChunkWriter;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -690,7 +691,15 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
boolean[] timeDuplicateInfo,
BitMap allValueColDeletedMap,
int maxNumberOfPointsInPage) {
+ // No duplicate timestamps, deleted rows, or all-null rows need filtering,
so avoid metadata
+ // checks.
AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
+ boolean hasTimeDeleted = alignedWorkingListForFlush.getTimeColDeletedMap()
!= null;
+ if (timeDuplicateInfo == null && allValueColDeletedMap == null &&
!hasTimeDeleted) {
+ handleEncodingWithoutDeletedMeasurementsFastPath(ioTaskQueue,
chunkRange);
+ return;
+ }
+
List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new
Pair[dataTypes.size()];
for (List<Integer> pageRange : chunkRange) {
@@ -840,6 +849,242 @@ public class AlignedWritableMemChunk extends
AbstractWritableMemChunk {
}
}
+ private void handleEncodingWithoutDeletedMeasurementsFastPath(
+ BlockingQueue<Object> ioTaskQueue, List<List<Integer>> chunkRange) {
+ AlignedTVList alignedWorkingListForFlush = (AlignedTVList)
workingListForFlush;
+ List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
+ // Sorting rearranges timestamps in place, so sorted row offsets map
directly to these arrays.
+ List<long[]> timestamps = alignedWorkingListForFlush.getTimestamps();
+ List<List<Object>> values = alignedWorkingListForFlush.getValues();
+ List<List<BitMap>> bitMaps = alignedWorkingListForFlush.getBitMaps();
+ BitMap segmentMovedMap = alignedWorkingListForFlush.getSegmentMovedMap();
+ boolean[] nulls = new boolean[ARRAY_SIZE];
+ for (List<Integer> pageRange : chunkRange) {
+ AlignedChunkWriterImpl alignedChunkWriter =
+ new AlignedChunkWriterImpl(schemaList, encryptParameter);
+ for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum++) {
+ int pageStart = pageRange.get(pageNum * 2);
+ int pageEnd = pageRange.get(pageNum * 2 + 1);
+ for (int segmentStart = pageStart; segmentStart <= pageEnd; ) {
+ int arrayIndex = segmentStart / ARRAY_SIZE;
+ int arrayOffset = segmentStart % ARRAY_SIZE;
+ int segmentEnd = Math.min(pageEnd, (arrayIndex + 1) * ARRAY_SIZE -
1);
+ int pointsInSegment = segmentEnd - segmentStart + 1;
+ long[] timestampArray = timestamps.get(arrayIndex);
+ boolean segmentMoved = segmentMovedMap != null &&
segmentMovedMap.isMarked(arrayIndex);
+ for (int columnIndex = 0; columnIndex < dataTypes.size();
columnIndex++) {
+ TSDataType tsDataType = dataTypes.get(columnIndex);
+ ValueChunkWriter valueChunkWriter =
+ alignedChunkWriter.getValueChunkWriterByIndex(columnIndex);
+ List<Object> columnValues = values.get(columnIndex);
+ Object valueArray = columnValues == null ? null :
columnValues.get(arrayIndex);
+ if (columnValues == null || (!segmentMoved && valueArray == null))
{
+ // An unmoved, unmaterialized segment is all null. A moved
segment can reference
+ // non-null values in another array, unless the entire column is
absent.
+ valueChunkWriter.getPageWriter().writeNull(pointsInSegment);
+ } else if (!segmentMoved) {
+ BitMap columnBitMap = null;
+ if (bitMaps != null && bitMaps.get(columnIndex) != null) {
+ columnBitMap = bitMaps.get(columnIndex).get(arrayIndex);
+ }
+ if (columnBitMap == null) {
+ writeValuesFromArray(
+ valueChunkWriter,
+ tsDataType,
+ timestampArray,
+ valueArray,
+ nulls,
+ arrayOffset,
+ pointsInSegment);
+ } else {
+ writeValuesFromArray(
+ valueChunkWriter,
+ tsDataType,
+ timestampArray,
+ valueArray,
+ columnBitMap,
+ arrayOffset,
+ pointsInSegment);
+ }
+ } else {
+ for (int sortedRowIndex = segmentStart;
+ sortedRowIndex <= segmentEnd;
+ sortedRowIndex++) {
+ int valueIndex =
alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
+ int valueArrayIndex = valueIndex / ARRAY_SIZE;
+ int valueElementIndex = valueIndex % ARRAY_SIZE;
+ Object sourceValueArray = columnValues.get(valueArrayIndex);
+ boolean isNull =
+ sourceValueArray == null
+ || alignedWorkingListForFlush.isNullValue(valueIndex,
columnIndex);
+ writeValueFromArray(
+ valueChunkWriter,
+ tsDataType,
+ timestampArray[arrayOffset + sortedRowIndex -
segmentStart],
+ sourceValueArray,
+ valueElementIndex,
+ isNull);
+ }
+ }
+ }
+ alignedChunkWriter.write(timestampArray, pointsInSegment,
arrayOffset);
+ segmentStart = segmentEnd + 1;
+ }
+ }
+ alignedChunkWriter.sealCurrentPage();
+ alignedChunkWriter.clearPageWriter();
+ try {
+ ioTaskQueue.put(alignedChunkWriter);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private void writeValueFromArray(
+ ValueChunkWriter valueChunkWriter,
+ TSDataType dataType,
+ long time,
+ Object valueArray,
+ int elementIndex,
+ boolean isNull) {
+ switch (dataType) {
+ case BOOLEAN ->
+ valueChunkWriter.write(
+ time,
+ !isNull && valueArray != null && ((boolean[])
valueArray)[elementIndex],
+ isNull);
+ case INT32, DATE ->
+ valueChunkWriter.write(
+ time, isNull || valueArray == null ? 0 : ((int[])
valueArray)[elementIndex], isNull);
+ case INT64, TIMESTAMP ->
+ valueChunkWriter.write(
+ time, isNull || valueArray == null ? 0 : ((long[])
valueArray)[elementIndex], isNull);
+ case FLOAT ->
+ valueChunkWriter.write(
+ time,
+ isNull || valueArray == null ? 0 : ((float[])
valueArray)[elementIndex],
+ isNull);
+ case DOUBLE ->
+ valueChunkWriter.write(
+ time,
+ isNull || valueArray == null ? 0 : ((double[])
valueArray)[elementIndex],
+ isNull);
+ case TEXT, STRING, BLOB, OBJECT ->
+ valueChunkWriter.write(
+ time,
+ isNull || valueArray == null ? null : ((Binary[])
valueArray)[elementIndex],
+ isNull);
+ case VECTOR, UNKNOWN -> throw new
UnSupportedDataTypeException(UNSUPPORTED_TYPE + dataType);
+ }
+ }
+
+ private void writeValuesFromArray(
+ ValueChunkWriter valueChunkWriter,
+ TSDataType dataType,
+ long[] timestamps,
+ Object valueArray,
+ boolean[] nulls,
+ int arrayOffset,
+ int pointsInSegment) {
+ switch (dataType) {
+ case BOOLEAN ->
+ valueChunkWriter.write(
+ timestamps, (boolean[]) valueArray, nulls, pointsInSegment,
arrayOffset);
+ case INT32, DATE ->
+ valueChunkWriter.write(
+ timestamps, (int[]) valueArray, nulls, pointsInSegment,
arrayOffset);
+ case INT64, TIMESTAMP ->
+ valueChunkWriter.write(
+ timestamps, (long[]) valueArray, nulls, pointsInSegment,
arrayOffset);
+ case FLOAT ->
+ valueChunkWriter.write(
+ timestamps, (float[]) valueArray, nulls, pointsInSegment,
arrayOffset);
+ case DOUBLE ->
+ valueChunkWriter.write(
+ timestamps, (double[]) valueArray, nulls, pointsInSegment,
arrayOffset);
+ case TEXT, STRING, BLOB, OBJECT ->
+ valueChunkWriter.write(
+ timestamps, (Binary[]) valueArray, nulls, pointsInSegment,
arrayOffset);
+ case VECTOR, UNKNOWN -> throw new
UnSupportedDataTypeException(UNSUPPORTED_TYPE + dataType);
+ }
+ }
+
+ private void writeValuesFromArray(
+ ValueChunkWriter valueChunkWriter,
+ TSDataType dataType,
+ long[] timestamps,
+ Object valueArray,
+ BitMap bitMap,
+ int arrayOffset,
+ int pointsInSegment) {
+ // Keep null detection in the bitmap so the page writer can skip encoding
and statistics
+ // updates without materializing a temporary boolean array.
+ switch (dataType) {
+ case BOOLEAN ->
+ valueChunkWriter
+ .getPageWriter()
+ .write(
+ timestamps,
+ (boolean[]) valueArray,
+ bitMap,
+ arrayOffset,
+ pointsInSegment,
+ arrayOffset);
+ case INT32, DATE ->
+ valueChunkWriter
+ .getPageWriter()
+ .write(
+ timestamps,
+ (int[]) valueArray,
+ bitMap,
+ arrayOffset,
+ pointsInSegment,
+ arrayOffset);
+ case INT64, TIMESTAMP ->
+ valueChunkWriter
+ .getPageWriter()
+ .write(
+ timestamps,
+ (long[]) valueArray,
+ bitMap,
+ arrayOffset,
+ pointsInSegment,
+ arrayOffset);
+ case FLOAT ->
+ valueChunkWriter
+ .getPageWriter()
+ .write(
+ timestamps,
+ (float[]) valueArray,
+ bitMap,
+ arrayOffset,
+ pointsInSegment,
+ arrayOffset);
+ case DOUBLE ->
+ valueChunkWriter
+ .getPageWriter()
+ .write(
+ timestamps,
+ (double[]) valueArray,
+ bitMap,
+ arrayOffset,
+ pointsInSegment,
+ arrayOffset);
+ case TEXT, STRING, BLOB, OBJECT ->
+ valueChunkWriter
+ .getPageWriter()
+ .write(
+ timestamps,
+ (Binary[]) valueArray,
+ bitMap,
+ arrayOffset,
+ pointsInSegment,
+ arrayOffset);
+ case VECTOR, UNKNOWN -> throw new
UnSupportedDataTypeException(UNSUPPORTED_TYPE + dataType);
+ }
+ }
+
private void handleEncodingWithDeletedMeasurements(
BlockingQueue<Object> ioTaskQueue,
List<List<Integer>> chunkRange,
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 28ec348a1a8..d2176c6891e 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
@@ -150,6 +150,9 @@ public abstract class AlignedTVList extends TVList {
// constructed after deletion
BitMap timeColDeletedMap;
+ // A marked bit means sorting changed at least one value index in the
corresponding segment.
+ private BitMap segmentMovedMap;
+
protected int timeDeletedCnt = 0;
private final AlignedTVList outer = this;
@@ -231,6 +234,7 @@ public abstract class AlignedTVList extends TVList {
AlignedTVList alignedTvList = AlignedTVList.newAlignedList(new
ArrayList<>(dataTypeList));
alignedTvList.timestamps = this.timestamps;
alignedTvList.indices = this.indices;
+ alignedTvList.segmentMovedMap = cloneSegmentMovedMap();
alignedTvList.values = values;
alignedTvList.bitMaps = bitMaps;
alignedTvList.rowCount = this.rowCount;
@@ -260,6 +264,7 @@ public abstract class AlignedTVList extends TVList {
cloneList.values = this.values;
cloneList.bitMaps = this.bitMaps;
cloneList.timeColDeletedMap = this.timeColDeletedMap;
+ cloneList.segmentMovedMap = cloneSegmentMovedMap();
cloneList.materializedValueArrayCounts =
Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
@@ -272,6 +277,7 @@ public abstract class AlignedTVList extends TVList {
AlignedTVList cloneList = AlignedTVList.newAlignedList(new
ArrayList<>(dataTypes));
cloneAs(cloneList);
cloneColumnDataTo(cloneList, null);
+ cloneList.segmentMovedMap = cloneSegmentMovedMap();
cloneList.materializedValueArrayCounts =
Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
@@ -298,6 +304,7 @@ public abstract class AlignedTVList extends TVList {
}
cloneAs(cloneList);
cloneColumnDataTo(cloneList, retainedColumns);
+ cloneList.segmentMovedMap = cloneSegmentMovedMap();
return prepareMovePlan(cloneList, retainedColumns);
}
@@ -509,6 +516,7 @@ public abstract class AlignedTVList extends TVList {
if (sorted) {
if (rowCount > 1 && timestamp < getTime(rowCount - 2)) {
sorted = false;
+ segmentMovedMap = null;
} else {
seqRowCount++;
}
@@ -824,6 +832,37 @@ public abstract class AlignedTVList extends TVList {
return values;
}
+ public BitMap getSegmentMovedMap() {
+ return segmentMovedMap;
+ }
+
+ protected void updateSegmentMovedMap() {
+ segmentMovedMap = null;
+ if (indices == null) {
+ return;
+ }
+
+ int segmentCount = (rowCount + ARRAY_SIZE - 1) / ARRAY_SIZE;
+ for (int segmentIndex = 0; segmentIndex < segmentCount; segmentIndex++) {
+ int segmentStart = segmentIndex * ARRAY_SIZE;
+ int segmentEnd = Math.min(segmentStart + ARRAY_SIZE, rowCount);
+ int[] indexArray = indices.get(segmentIndex);
+ for (int rowIndex = segmentStart; rowIndex < segmentEnd; rowIndex++) {
+ if (indexArray[rowIndex - segmentStart] != rowIndex) {
+ if (segmentMovedMap == null) {
+ segmentMovedMap = BitMap.createBitMapDynamically(segmentCount);
+ }
+ segmentMovedMap.mark(segmentIndex);
+ break;
+ }
+ }
+ }
+ }
+
+ private BitMap cloneSegmentMovedMap() {
+ return segmentMovedMap == null ? null : segmentMovedMap.clone();
+ }
+
public List<TSDataType> getTsDataTypes() {
return dataTypes;
}
@@ -1108,6 +1147,7 @@ public abstract class AlignedTVList extends TVList {
}
}
materializedBitmapMemoryCost = 0;
+ segmentMovedMap = null;
}
@Override
@@ -1115,6 +1155,9 @@ public abstract class AlignedTVList extends TVList {
if (indices != null) {
indices.add((int[]) getPrimitiveArraysByType(TSDataType.INT32));
}
+ if (segmentMovedMap != null) {
+ segmentMovedMap.extend(segmentMovedMap.getSize() + 1);
+ }
for (int i = 0; i < dataTypes.size(); i++) {
List<Object> columnValues = values.get(i);
if (columnValues == null) {
@@ -1186,6 +1229,9 @@ public abstract class AlignedTVList extends TVList {
int idx = start;
updateMinMaxTimeAndSorted(time, start, end);
+ if (!sorted) {
+ segmentMovedMap = null;
+ }
while (idx < end) {
int inputRemaining = end - idx;
@@ -1537,6 +1583,9 @@ public abstract class AlignedTVList extends TVList {
}
}
}
+ if (segmentMovedMap != null) {
+ size += segmentMovedMap.ramBytesUsed();
+ }
return size;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
index c4f5174d195..c6d768b74ad 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
@@ -40,6 +40,7 @@ public class BackAlignedTVList extends QuickAlignedTVList {
if (!sorted) {
policy.backwardSort(timestamps, rowCount);
policy.clearTmp();
+ updateSegmentMovedMap();
}
sorted = true;
seqRowCount = rowCount;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
index 9ea50a06246..fcec76e1c6b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
@@ -38,6 +38,7 @@ public class QuickAlignedTVList extends AlignedTVList {
public synchronized int sort() {
if (!sorted) {
policy.qsort(0, rowCount - 1);
+ updateSegmentMovedMap();
}
sorted = true;
seqRowCount = rowCount;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
index c39297d59b6..b37ed9d0d7b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
@@ -36,9 +36,10 @@ public class TimAlignedTVList extends AlignedTVList {
@Override
public synchronized int sort() {
- policy.checkSortedTimestampsAndIndices();
if (!sorted) {
+ policy.checkSortedTimestampsAndIndices();
policy.sort(0, rowCount);
+ updateSegmentMovedMap();
}
policy.clearSortedValue();
policy.clearSortedTime();
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 3c55c65517f..8df5e71c836 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
@@ -27,15 +27,28 @@ import org.apache.iotdb.db.utils.constant.TestConstant;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.ChunkMetadata;
+import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.tsfile.read.TsFileReader;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.RowRecord;
+import org.apache.tsfile.read.expression.QueryExpression;
+import org.apache.tsfile.read.query.dataset.QueryDataSet;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.write.chunk.IChunkWriter;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.apache.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.apache.tsfile.write.writer.TsFileIOWriter;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -43,6 +56,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
+import static
org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -198,6 +212,182 @@ public class MemTableFlushTaskTest {
assertFalse(ioTaskQueue.isEmpty());
}
+ @Test
+ public void
testAlignedFastPathKeepsPagesAndValuesAlignedAfterPartialSegmentSort()
+ throws IOException, InterruptedException {
+ // Covers one moved segment and several untouched segments across multiple
logical pages.
+ int rowCount = 10_000;
+ List<IMeasurementSchema> schemas =
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT64, TSEncoding.PLAIN),
+ new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN));
+ AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas,
false);
+ String alignedFilePath =
TestConstant.OUTPUT_DATA_DIR.concat("testAlignedFastPath.tsfile");
+
+ try {
+ for (int index = 0; index < rowCount; index++) {
+ long time = index;
+ if (index == 100) {
+ time = 101;
+ } else if (index == 101) {
+ time = 100;
+ }
+ memChunk.putAlignedRow(time, new Object[] {time, time * 10});
+ }
+ memChunk.sortTvListForFlush();
+
+ BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>();
+ memChunk.encodeWorkingAlignedTVList(ioTaskQueue, rowCount, 1024);
+ 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) {
+ ((IChunkWriter) task).writeToFileWriter(alignedWriter);
+ }
+ }
+ alignedWriter.endChunkGroup();
+ alignedWriter.endFile();
+ }
+
+ try (TsFileSequenceReader sequenceReader = new
TsFileSequenceReader(alignedFilePath);
+ TsFileReader fileReader = new TsFileReader(sequenceReader)) {
+ QueryDataSet dataSet =
+ fileReader.query(
+ QueryExpression.create(
+ Arrays.asList(new Path("root.d", "s0", false), new
Path("root.d", "s1", false)),
+ null));
+ int index = 0;
+ while (dataSet.hasNext()) {
+ RowRecord row = dataSet.next();
+ assertEquals(index, row.getTimestamp());
+ assertEquals((long) index, row.getFields().get(0).getLongV());
+ assertEquals((long) index * 10, row.getFields().get(1).getLongV());
+ index++;
+ }
+ assertEquals(rowCount, index);
+ }
+ } finally {
+ memChunk.release();
+ }
+ }
+
+ @Test
+ public void testAlignedFastPathEncodesUnmaterializedSegments() throws
Exception {
+ // Exercise all six value representations with null/dense/null segments,
partial nulls, an
+ // entirely empty column, and page/chunk boundaries inside backing arrays.
+ checkUnmaterializedSegments(false);
+ }
+
+ @Test
+ public void
testAlignedFastPathDoesNotSkipValuesMovedIntoUnmaterializedSegments()
+ throws Exception {
+ // Sorting swaps rows between a null array and a materialized array. A
null array at the
+ // sorted segment's offset must not hide a non-null value mapped from the
other segment.
+ checkUnmaterializedSegments(true);
+ }
+
+ private void checkUnmaterializedSegments(boolean moved) throws Exception {
+ int rowCount = ARRAY_SIZE * 12 + 7;
+ int movedRow = ARRAY_SIZE + 1;
+ if (movedRow % 5 == 0) {
+ movedRow++;
+ }
+ List<TSDataType> types =
+ Arrays.asList(
+ TSDataType.BOOLEAN,
+ TSDataType.INT32,
+ TSDataType.INT64,
+ TSDataType.FLOAT,
+ TSDataType.DOUBLE,
+ TSDataType.TEXT);
+ List<IMeasurementSchema> schemas = new ArrayList<>();
+ schemas.add(new MeasurementSchema("anchor", TSDataType.INT64,
TSEncoding.PLAIN));
+ for (int column = 0; column < types.size(); column++) {
+ schemas.add(new MeasurementSchema("s" + column, types.get(column),
TSEncoding.PLAIN));
+ }
+ schemas.add(new MeasurementSchema("empty", TSDataType.INT64,
TSEncoding.PLAIN));
+ AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true);
+ Object[][] expected = new Object[rowCount][types.size()];
+ String path = TestConstant.OUTPUT_DATA_DIR.concat("unmaterialized-" +
moved + ".tsfile");
+ try {
+ for (int row = 0; row < rowCount; row++) {
+ int time = row;
+ if (moved && row == 1) {
+ time = movedRow;
+ } else if (moved && row == movedRow) {
+ time = 1;
+ }
+ Object[] values = new Object[schemas.size()];
+ values[0] = (long) time;
+ // Every third segment stays unmaterialized, and other segments also
contain nulls.
+ if (row / ARRAY_SIZE % 3 != 0 && row % 5 != 0) {
+ Object[] typedValues = {
+ time % 2 == 0,
+ time,
+ (long) time,
+ time + 0.5f,
+ time + 0.25d,
+ new Binary("value-" + time, StandardCharsets.UTF_8)
+ };
+ for (int column = 0; column < types.size(); column++) {
+ values[column + 1] = typedValues[column];
+ expected[time][column] = typedValues[column];
+ }
+ }
+ chunk.putAlignedRow(time, values);
+ }
+ // Verify that the test actually exercises lazy null arrays, not merely
marked bitmaps.
+ assertTrue(chunk.getWorkingTVList().getValues().get(1).get(0) == null);
+ assertTrue(chunk.getWorkingTVList().getValues().get(1).get(1) != null);
+ if (moved) {
+ assertTrue(expected[1][0] != null);
+ }
+ chunk.sortTvListForFlush();
+ BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
+ chunk.encodeWorkingAlignedTVList(queue, ARRAY_SIZE * 5 + 3, ARRAY_SIZE +
3);
+ try (TsFileIOWriter fileWriter = new TsFileIOWriter(new File(path))) {
+
fileWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d"));
+ Object task;
+ while ((task = queue.poll()) != null) {
+ if (task instanceof IChunkWriter chunkWriter) {
+ chunkWriter.writeToFileWriter(fileWriter);
+ }
+ }
+ fileWriter.endChunkGroup();
+ fileWriter.endFile();
+ }
+ try (TsFileSequenceReader sequence = new TsFileSequenceReader(path);
+ TsFileReader reader = new TsFileReader(sequence)) {
+ List<Path> paths = new ArrayList<>();
+ for (IMeasurementSchema schema : schemas) {
+ paths.add(new Path("root.d", schema.getMeasurementName(), false));
+ }
+ QueryDataSet data = reader.query(QueryExpression.create(paths, null));
+ int row = 0;
+ while (data.hasNext()) {
+ RowRecord record = data.next();
+ assertEquals(row, record.getTimestamp());
+ assertEquals((long) row, record.getFields().get(0).getLongV());
+ for (int column = 0; column < types.size(); column++) {
+ assertEquals(
+ expected[row][column],
+ record.getFields().get(column + 1) == null
+ ? null
+ : record.getFields().get(column +
1).getObjectValue(types.get(column)));
+ }
+ assertTrue(
+ record.getFields().get(schemas.size() - 1) == null
+ || record.getFields().get(schemas.size() - 1).getDataType()
== null);
+ row++;
+ }
+ assertEquals(rowCount, row);
+ }
+ } finally {
+ chunk.release();
+ }
+ }
+
private TrackingAlignedWritableMemChunk createTrackingAlignedMemChunk() {
List<IMeasurementSchema> schemas =
new ArrayList<>(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
index aceb2e23420..708f2113cdf 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
@@ -113,6 +113,38 @@ public class AlignedTVListTest {
}
}
+ @Test
+ public void testSegmentMovedMapOnlyMarksMovedSegments() {
+ AlignedTVList tvList =
AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64));
+ int rowCount = ARRAY_SIZE + 3;
+ for (int i = 0; i < rowCount; i++) {
+ // Keep the first segment ordered and introduce an inversion only in the
second segment.
+ long time =
+ i < ARRAY_SIZE
+ ? i
+ : (i == ARRAY_SIZE
+ ? ARRAY_SIZE + 1L
+ : i == ARRAY_SIZE + 1 ? ARRAY_SIZE + 2L : ARRAY_SIZE);
+ tvList.putAlignedValue(time, new Object[] {(long) i});
+ }
+
+ tvList.sort();
+
+ Assert.assertNotNull(tvList.getSegmentMovedMap());
+ Assert.assertTrue(tvList.getSegmentMovedMap().isMarked(1));
+ for (int segmentIndex = 0; segmentIndex < 2; segmentIndex++) {
+ boolean moved = false;
+ int segmentStart = segmentIndex * ARRAY_SIZE;
+ int segmentEnd = Math.min(segmentStart + ARRAY_SIZE, rowCount);
+ for (int rowIndex = segmentStart; rowIndex < segmentEnd; rowIndex++) {
+ moved |= tvList.getValueIndex(rowIndex) != rowIndex;
+ }
+ Assert.assertEquals(moved,
tvList.getSegmentMovedMap().isMarked(segmentIndex));
+ }
+ Assert.assertEquals(0, tvList.getValueIndex(0));
+ Assert.assertEquals(ARRAY_SIZE + 2, tvList.getValueIndex(ARRAY_SIZE));
+ }
+
@Test
public void testAlignedTVLists() {
List<TSDataType> dataTypes = new ArrayList<>();
diff --git a/pom.xml b/pom.xml
index 90ac0c70ce7..7d4b3d74a42 100644
--- a/pom.xml
+++ b/pom.xml
@@ -146,7 +146,7 @@
<thrift.version>0.24.0</thrift.version>
<xz.version>1.9</xz.version>
<zstd-jni.version>1.5.6-3</zstd-jni.version>
- <tsfile.version>2.4.1-260806-SNAPSHOT</tsfile.version>
+ <tsfile.version>2.4.1-260909-SNAPSHOT</tsfile.version>
<i18n.locale>en</i18n.locale>
<tsfile.locale.opt/>
</properties>
