This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch optimize_flush_encode in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6a6b0840a3243404cc5a4ec275b428e3193ba903 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 9 11:40:35 2026 +0800 use new tsfile interfaces --- .../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 +- 8 files changed, 521 insertions(+), 2 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 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>
