This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8608877e1c95e656a8db7904280d5a364f4d3c73 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 22 10:30:00 2026 +0800 tmp --- .../it/insertquery/IoTDBFlatDeviceIT.java | 57 ++++++ .../impl/ReadPointCompactionPerformer.java | 15 +- .../execute/utils/FlatDeviceCompactionUtils.java | 13 +- .../utils/writer/AbstractCompactionWriter.java | 99 +++++++---- .../writer/ReadPointCrossCompactionWriter.java | 22 ++- .../memtable/AlignedWritableMemChunk.java | 36 ++-- .../modification/FlatDeviceDeletionEntry.java | 12 ++ .../tsfile/FlatDeviceAlignedChunkWriter.java | 111 ++++++++++++ .../db/utils/datastructure/AlignedTVList.java | 68 +++++--- .../utils/datastructure/AlignedTVListFactory.java | 9 + .../datastructure/ColumnSortedAlignedTVList.java | 26 ++- .../datastructure/MemPointIteratorFactory.java | 4 + .../MergeSortMultiAlignedTVListIterator.java | 12 +- .../datastructure/MultiAlignedTVListIterator.java | 21 ++- .../CompactionTableModelTestFileWriter.java | 18 ++ .../tablemodel/FlatDeviceCompactionTest.java | 136 ++++++++++++++- .../memtable/FlatDeviceMemTableDeletionTest.java | 31 +++- .../memtable/FlatDeviceMemTableTest.java | 96 ++++++++++- .../ColumnSortedAlignedTVListTest.java | 192 ++++++++++++++++++++- 19 files changed, 853 insertions(+), 125 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java index 67fed2339e9..50b6e9cdc8d 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java @@ -216,6 +216,63 @@ public class IoTDBFlatDeviceIT { } } + // TAG-first memory and flush order restarts time for each device. SQL readers must still merge + // by time, apply range/limit filters correctly and preserve updates across memory and disk. + @Test + public void testTagFirstTimeRegressionAcrossFlush() throws SQLException { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + for (String tag : List.of("b", "a")) { + for (long time : new long[] {100, 1, 50}) { + statement.execute( + "INSERT INTO sensor(time, tag_a, tag_b, value) VALUES (" + + time + + ", '" + + tag + + "', 'x', " + + time + + ")"); + } + } + List<String> expected = + new ArrayList<>( + List.of( + "1,a,x,null,1.0", + "1,b,x,null,1.0", + "50,a,x,null,50.0", + "50,b,x,null,50.0", + "100,a,x,null,100.0", + "100,b,x,null,100.0")); + for (int phase = 0; phase < 4; phase++) { + assertDeletionRows(statement, expected); + assertRows( + statement.executeQuery( + "SELECT time, tag_a, tag_b, attr, value FROM sensor" + + " WHERE time >= 2 AND time <= 50 ORDER BY time, tag_a, tag_b"), + expected.subList(2, 4)); + assertRows( + statement.executeQuery( + "SELECT time, tag_a, tag_b, attr, value FROM sensor" + + " ORDER BY time DESC, tag_a DESC LIMIT 2 OFFSET 2"), + List.of(expected.get(3), expected.get(2))); + if (phase == 1) { + statement.execute( + "INSERT INTO sensor(time, tag_a, tag_b, value) VALUES (50, 'b', 'x', 500)"); + expected.set(3, "50,b,x,null,500.0"); + } else { + statement.execute("FLUSH"); + } + } + statement.execute("DELETE FROM sensor WHERE time = 50"); + expected.remove(3); + expected.remove(2); + assertDeletionRows(statement, expected); + statement.execute("FLUSH"); + assertDeletionRows(statement, expected); + } + } + // Covers Tablet writes whose rows span multiple logical devices and verifies the shared physical // group preserves time-plus-TAG ordering across flush. @Test diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java index 0803f1f8aa0..80f6cd78bca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java @@ -45,6 +45,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimato import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator.RepairUnsortedFileCompactionEstimator; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.read.control.QueryResourceManager; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.utils.EncryptDBUtils; @@ -251,13 +252,17 @@ public class ReadPointCompactionPerformer compactionWriter.startChunkGroup( device, true, FlatDeviceCompactionUtils.isFlatDevice(tableSchema)); compactionWriter.setRowKeyColumnIndices(rowKeyColumnIndices); - measurementSchemas.add(0, timeSchema); compactionWriter.startMeasurement( TsFileConstant.TIME_COLUMN_ID, - new AlignedChunkWriterImpl( - measurementSchemas.remove(0), - measurementSchemas, - EncryptUtils.getEncryptParameter(getEncryptParameter())), + FlatDeviceCompactionUtils.isFlatDevice(tableSchema) + ? new FlatDeviceAlignedChunkWriter( + timeSchema, + measurementSchemas, + EncryptUtils.getEncryptParameter(getEncryptParameter())) + : new AlignedChunkWriterImpl( + timeSchema, + measurementSchemas, + EncryptUtils.getEncryptParameter(getEncryptParameter())), 0); writeWithReader(compactionWriter, dataBlockReader, device, 0, true); compactionWriter.endMeasurement(0); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java index 57ca4e45dbb..791298d3c46 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/FlatDeviceCompactionUtils.java @@ -97,8 +97,17 @@ public class FlatDeviceCompactionUtils { } public static void updateResource( - TsFileResource resource, IDeviceID physicalDevice, List<Integer> tagIndices, TsBlock block) { - for (int row = 0; row < block.getPositionCount(); row++) { + TsFileResource resource, IDeviceID physicalDevice, TsBlock block) { + updateResource(resource, physicalDevice, block, 0, block.getPositionCount()); + } + + public static void updateResource( + TsFileResource resource, + IDeviceID physicalDevice, + TsBlock block, + int start, + int count) { + for (int row = start; row < start + count; row++) { resource.updateTableStartTime(physicalDevice.getTableName(), block.getTimeByIndex(row)); resource.updateTableEndTime(physicalDevice.getTableName(), block.getTimeByIndex(row)); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java index c5e873975c6..15ecadd62b4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java @@ -25,16 +25,15 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionLastTimeCheckFailedException; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CompactionTaskSummary; -import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.batch.utils.FollowingBatchCompactionAlignedChunkWriter; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.AlignedPageElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; -import org.apache.iotdb.db.storageengine.dataregion.read.reader.common.AlignedPriorityMergeReader; import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.write.PageException; @@ -46,7 +45,7 @@ import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.block.TsBlock; -import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.TsPrimitiveType; import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; import org.apache.tsfile.write.chunk.ChunkWriterImpl; @@ -57,14 +56,12 @@ import org.apache.tsfile.write.writer.TsFileIOWriter; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.stream.Collectors; public abstract class AbstractCompactionWriter implements AutoCloseable { protected int subTaskNum = IoTDBDescriptor.getInstance().getConfig().getSubCompactionTaskNum(); @@ -136,15 +133,47 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { } Set<String> devices = flatDeviceLogicalDevices.computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()); - IPointReader rows = block.getTsBlockAlignedRowIterator(); - while (rows.hasNextTimeValuePair()) { - devices.add( - FlatDeviceCompactionUtils.getLogicalDevice(deviceId, tagIndices, rows.nextTimeValuePair()) - .toString()); - } + recordFlatDeviceLogicalDevices(block, tagIndices, devices); flatDeviceRowCounts.merge(deviceId.getTableName(), (long) block.getPositionCount(), Long::sum); } + protected void recordFlatDeviceLogicalDevices( + TsBlock block, List<Integer> tagIndices, Set<String> devices) { + recordFlatDeviceLogicalDevices(block, 0, block.getPositionCount(), tagIndices, devices); + } + + protected void recordFlatDeviceLogicalDevices( + TsBlock block, int start, int count, List<Integer> tagIndices, Set<String> devices) { + int end = start + count; + for (int row = start; row < end; row++) { + boolean newDevice = row == start; + for (int tag : tagIndices) { + Column column = block.getColumn(tag); + if (newDevice + || column.isNull(row) != column.isNull(row - 1) + || (!column.isNull(row) && !column.getBinary(row).equals(column.getBinary(row - 1)))) { + newDevice = true; + break; + } + } + if (!newDevice) { + continue; + } + // TAG-first rows keep each logical device contiguous. Materialize only its TAGs once + // per run instead of boxing every FIELD on every row for the file statistics. + String[] segments = new String[tagIndices.size() + 1]; + segments[0] = deviceId.getTableName(); + for (int tag = 0; tag < tagIndices.size(); tag++) { + Column column = block.getColumn(tagIndices.get(tag)); + segments[tag + 1] = + column.isNull(row) + ? null + : column.getBinary(row).getStringValue(TSFileConfig.STRING_CHARSET); + } + devices.add(IDeviceID.Factory.DEFAULT_FACTORY.create(segments).toString()); + } + } + protected void writeFlatDeviceStatistics(TsFileIOWriter writer) { flatDeviceLogicalDevices.forEach( (table, devices) -> @@ -153,32 +182,41 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { } protected List<Integer> rowKeyColumnIndices = Collections.emptyList(); - private TimeValuePair lastRowKey; + private Binary[] lastRowTags; + private long lastRowTime; + private boolean hasLastRowKey; public void setRowKeyColumnIndices(List<Integer> rowKeyColumnIndices) { this.rowKeyColumnIndices = rowKeyColumnIndices; - lastRowKey = null; + lastRowTags = new Binary[rowKeyColumnIndices.size()]; + hasLastRowKey = false; } protected void checkFlatDeviceRowOrder(TsBlock block) throws IOException { if (rowKeyColumnIndices.isEmpty()) { return; } - List<TSDataType> types = - Arrays.stream(block.getValueColumns()) - .map(Column::getDataType) - .collect(Collectors.toList()); - IPointReader rows = block.getTsBlockAlignedRowIterator(); - while (rows.hasNextTimeValuePair()) { - TimeValuePair row = rows.nextTimeValuePair(); - if (lastRowKey != null - && AlignedPriorityMergeReader.compareFlatDeviceRowKey( - lastRowKey, row, types, rowKeyColumnIndices) - >= 0) { - throw new CompactionLastTimeCheckFailedException( - deviceId.toString(), row.getTimestamp(), lastRowKey.getTimestamp()); + for (int row = 0; row < block.getPositionCount(); row++) { + int comparison = 0; + for (int tag = 0; tag < rowKeyColumnIndices.size(); tag++) { + Column column = block.getColumn(rowKeyColumnIndices.get(tag)); + Binary current = column.isNull(row) ? null : column.getBinary(row); + Binary previous = lastRowTags[tag]; + if (comparison == 0) { + comparison = + previous == null + ? (current == null ? 0 : -1) + : (current == null ? 1 : previous.compareTo(current)); + } + // Keep only TAG references and time across blocks, never a boxed row or its FIELDs. + lastRowTags[tag] = current; + } + long time = block.getTimeByIndex(row); + if (hasLastRowKey && (comparison > 0 || (comparison == 0 && lastRowTime >= time))) { + throw new CompactionLastTimeCheckFailedException(deviceId.toString(), time, lastRowTime); } - lastRowKey = row; + lastRowTime = time; + hasLastRowKey = true; } } @@ -329,13 +367,16 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { } protected long estimateWrittenPointTotalSize(TsBlock tsBlock) { - int pointCount = tsBlock.getPositionCount(); + return estimateWrittenPointTotalSize(tsBlock, 0, tsBlock.getPositionCount()); + } + + protected long estimateWrittenPointTotalSize(TsBlock tsBlock, int start, int pointCount) { long size = (long) Long.BYTES * pointCount; Column[] columns = tsBlock.getValueColumns(); for (Column column : columns) { TSDataType dataType = column.getDataType(); if (dataType.isBinary()) { - for (int j = 0; j < pointCount; j++) { + for (int j = start; j < start + pointCount; j++) { if (column.isNull(j)) { continue; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java index 8d447619b2e..5051c1d68b9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.FlatDeviceCompactionUtils; @@ -26,6 +27,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.exe import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; import org.apache.iotdb.db.storageengine.dataregion.read.control.FileReaderManager; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.utils.EncryptDBUtils; @@ -38,12 +40,19 @@ import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.common.block.column.TimeColumn; import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.writer.TsFileIOWriter; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWriter { + private final Map<TsFileIOWriter, Map<String, Set<String>>> flatDevicesByFile = new HashMap<>(); + private final Map<TsFileIOWriter, Map<String, Long>> flatRowsByFile = new HashMap<>(); @TestOnly public ReadPointCrossCompactionWriter( @@ -63,7 +72,10 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite @Override public void write(TsBlock tsBlock, int subTaskId) throws IOException { checkFlatDeviceRowOrder(tsBlock); - recordFlatDeviceRows(tsBlock, rowKeyColumnIndices); + if (!rowKeyColumnIndices.isEmpty()) { + writeFlatDeviceBlock(tsBlock, subTaskId); + return; + } TimeColumn timestamps = (TimeColumn) tsBlock.getTimeColumn(); Column[] columns = tsBlock.getValueColumns(); int batchSize = tsBlock.getPositionCount(); @@ -79,12 +91,8 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite synchronized (this) { // we need to synchronized here to avoid multi-thread competition in sub-task TsFileResource resource = targetResources.get(seqFileIndexArray[subTaskId]); - if (rowKeyColumnIndices.isEmpty()) { - resource.updateStartTime(deviceId, timestamps.getStartTime()); - resource.updateEndTime(deviceId, timestamps.getEndTime()); - } else { - FlatDeviceCompactionUtils.updateResource(resource, deviceId, rowKeyColumnIndices, tsBlock); - } + resource.updateStartTime(deviceId, timestamps.getStartTime()); + resource.updateEndTime(deviceId, timestamps.getEndTime()); } checkChunkSizeAndMayOpenANewChunk( targetFileWriters.get(seqFileIndexArray[subTaskId]), chunkWriter, subTaskId); 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 5c1658c69e6..1c04b805f6e 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 @@ -29,6 +29,7 @@ import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; @@ -43,7 +44,6 @@ import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.enums.TSDataType; -import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.read.filter.basic.Filter; import org.apache.tsfile.utils.Binary; @@ -57,7 +57,6 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import java.io.DataInputStream; import java.io.IOException; -import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; @@ -171,7 +170,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.list = !useColumnSortedList ? AlignedTVListFactory.createAlignedList(dataTypes) - : AlignedTVListFactory.createColumnSortedAlignedList(dataTypes, this.sortColumnIndices); + : AlignedTVListFactory.createFlatDeviceAlignedList(dataTypes, this.sortColumnIndices); this.sortedList = new ArrayList<>(); this.ignoreAllNullRows = !isTableModel; this.encryptParameter = encryptParameter; @@ -322,7 +321,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.list = !this.useColumnSortedList ? AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)) - : AlignedTVListFactory.createColumnSortedAlignedList( + : AlignedTVListFactory.createFlatDeviceAlignedList( new ArrayList<>(dataTypes), this.sortColumnIndices); this.dataTypes = this.list.getTsDataTypes(); } @@ -776,7 +775,13 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { @Override public IChunkWriter createIChunkWriter() { - return new AlignedChunkWriterImpl(getActiveSchemaList(), encryptParameter); + return createAlignedChunkWriter(getActiveSchemaList()); + } + + private AlignedChunkWriterImpl createAlignedChunkWriter(List<IMeasurementSchema> schemas) { + return useColumnSortedList + ? new FlatDeviceAlignedChunkWriter(schemas, encryptParameter) + : new AlignedChunkWriterImpl(schemas, encryptParameter); } private List<IMeasurementSchema> getActiveSchemaList() { @@ -810,10 +815,6 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { BlockingQueue<Object> ioTaskQueue, long maxNumberOfPointsInChunk, int maxNumberOfPointsInPage) { - if (useColumnSortedList) { - encodeFlatDevice(ioTaskQueue, maxNumberOfPointsInChunk, maxNumberOfPointsInPage); - return; - } List<IMeasurementSchema> activeSchemaList = getActiveSchemaList(); if (activeSchemaList.isEmpty() && ignoreAllNullRows) { return; @@ -935,8 +936,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes(); Pair<Integer, Integer>[] lastValidPointIndexForKeyDupCheck = new Pair[dataTypes.size()]; for (List<Integer> pageRange : chunkRange) { - AlignedChunkWriterImpl alignedChunkWriter = - new AlignedChunkWriterImpl(schemaList, encryptParameter); + AlignedChunkWriterImpl alignedChunkWriter = createAlignedChunkWriter(schemaList); for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) { for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) { // Pair of sorted row index (for key comparison) and original value index. @@ -1095,8 +1095,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList); Pair<Integer, Integer>[] lastValidPointIndexForKeyDupCheck = new Pair[activeSchemaList.size()]; for (List<Integer> pageRange : chunkRange) { - AlignedChunkWriterImpl alignedChunkWriter = - new AlignedChunkWriterImpl(activeSchemaList, encryptParameter); + AlignedChunkWriterImpl alignedChunkWriter = createAlignedChunkWriter(activeSchemaList); for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) { for (int columnIndex = 0; columnIndex < activeSchemaList.size(); columnIndex++) { int tvListColumnIndex = columnIndexList.get(columnIndex); @@ -1257,12 +1256,6 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { encodeInfo.maxNumberOfPointsInChunk, (encodeInfo.targetChunkSize / getAvgPointSizeOfLargestColumn())); - if (useColumnSortedList) { - encodeFlatDevice( - ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.maxNumberOfPointsInPage); - return; - } - if (TVLIST_SORT_THRESHOLD == 0) { encodeWorkingAlignedTVList( ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.maxNumberOfPointsInPage); @@ -1274,8 +1267,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { return; } - AlignedChunkWriterImpl alignedChunkWriter = - new AlignedChunkWriterImpl(activeSchemaList, encryptParameter); + AlignedChunkWriterImpl alignedChunkWriter = createAlignedChunkWriter(activeSchemaList); // create MergeSortAlignedTVListIterator. List<AlignedTVList> alignedTvLists = new ArrayList<>(sortedList); @@ -1305,7 +1297,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - alignedChunkWriter = new AlignedChunkWriterImpl(activeSchemaList, encryptParameter); + alignedChunkWriter = createAlignedChunkWriter(activeSchemaList); encodeInfo.reset(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java index ab117841034..9efe318841b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/FlatDeviceDeletionEntry.java @@ -17,6 +17,8 @@ */ package org.apache.iotdb.db.storageengine.dataregion.modification; +import org.apache.iotdb.db.utils.ModificationUtils; + import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; @@ -118,6 +120,16 @@ public class FlatDeviceDeletionEntry extends TableDeletionEntry { return getTableName().equals(deviceID.getTableName()); } + /** + * Flat-device time indexes contain only the table device, so TAG predicates cannot be evaluated + * while checking whether a file may contain affected data. Defer that match to row filtering. + */ + @Override + public boolean affects(IDeviceID deviceID, long startTime, long endTime) { + return affects(deviceID) + && ModificationUtils.overlap(getStartTime(), getEndTime(), startTime, endTime); + } + @Override public boolean affectsAll(IDeviceID deviceID) { return affects(deviceID) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/FlatDeviceAlignedChunkWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/FlatDeviceAlignedChunkWriter.java new file mode 100644 index 00000000000..4da14ef525d --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/FlatDeviceAlignedChunkWriter.java @@ -0,0 +1,111 @@ +/* + * 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.tsfile; + +import org.apache.iotdb.db.i18n.DataNodeMiscMessages; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.read.common.block.column.TimeColumn; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.chunk.ValueChunkWriter; +import org.apache.tsfile.write.schema.IMeasurementSchema; + +import java.util.List; + +/** Keeps aligned column encoding and page boundaries while allowing TAG-first timestamps. */ +public class FlatDeviceAlignedChunkWriter extends AlignedChunkWriterImpl { + public FlatDeviceAlignedChunkWriter( + List<IMeasurementSchema> schemas, EncryptParameter encryptParameter) { + super(schemas, encryptParameter); + } + + public FlatDeviceAlignedChunkWriter( + IMeasurementSchema timeSchema, + List<IMeasurementSchema> valueSchemas, + EncryptParameter encryptParameter) { + super(timeSchema, valueSchemas, encryptParameter); + } + + @Override + public void write(long[] times, int batchSize, int arrayOffset) { + valueIndex = 0; + // The ordinary batch time writer assumes its first/last timestamps are min/max. A TAG + // switch can move time backwards, so update statistics while encoding each timestamp. + // Values for this batch are already encoded: check page size only after all times are + // written, preserving alignment and avoiding extra pages at TAG boundaries. + for (int i = arrayOffset; i < arrayOffset + batchSize; i++) { + timeChunkWriter.write(times[i]); + } + finishTimeBatch(); + } + + /** + * Writes a range from the original block without materializing a region view. The regular aligned + * writer receives arrayOffset zero because its callers normally pass a whole block; flat cross + * compaction partitions one block when it switches target files. + */ + public void write(TimeColumn timeColumn, Column[] valueColumns, int batchSize, int arrayOffset) { + if (remainingPointsNumber < batchSize) { + int pointsHasWritten = (int) remainingPointsNumber; + batchWrite(timeColumn, valueColumns, pointsHasWritten, arrayOffset); + batchWrite( + timeColumn, valueColumns, batchSize - pointsHasWritten, arrayOffset + pointsHasWritten); + } else { + batchWrite(timeColumn, valueColumns, batchSize, arrayOffset); + } + } + + private void batchWrite( + TimeColumn timeColumn, Column[] valueColumns, int batchSize, int arrayOffset) { + long[] times = timeColumn.getTimes(); + valueIndex = 0; + for (Column column : valueColumns) { + ValueChunkWriter chunkWriter = valueChunkWriterList.get(valueIndex++); + switch (chunkWriter.getDataType()) { + case TEXT, STRING, BLOB, OBJECT -> + chunkWriter.write(times, column.getBinaries(), column.isNull(), batchSize, arrayOffset); + case DOUBLE -> + chunkWriter.write(times, column.getDoubles(), column.isNull(), batchSize, arrayOffset); + case BOOLEAN -> + chunkWriter.write(times, column.getBooleans(), column.isNull(), batchSize, arrayOffset); + case INT64, TIMESTAMP -> + chunkWriter.write(times, column.getLongs(), column.isNull(), batchSize, arrayOffset); + case INT32, DATE -> + chunkWriter.write(times, column.getInts(), column.isNull(), batchSize, arrayOffset); + case FLOAT -> + chunkWriter.write(times, column.getFloats(), column.isNull(), batchSize, arrayOffset); + case VECTOR, UNKNOWN -> + throw new UnsupportedOperationException( + String.format( + DataNodeMiscMessages.MISC_EXCEPTION_DATA_TYPE_S_IS_NOT_SUPPORTED_5D5C02E4, + chunkWriter.getDataType())); + } + } + write(times, batchSize, arrayOffset); + } + + private void finishTimeBatch() { + if (checkPageSizeAndMayOpenANewPage()) { + writePageToPageBuffer(); + } + remainingPointsNumber = timeChunkWriter.getRemainingPointNumberForCurrentPage(); + } +} 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 d0bc9bb67b8..0111159ff1a 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 @@ -225,10 +225,27 @@ public abstract class AlignedTVList extends TVList { // Default AlignedTVList implementations have no subtype-specific state to copy. } + protected void copyStateForFlushSortTo(AlignedTVList cloneList) { + copyAdditionalStateTo(cloneList); + } + + protected void copyStateForPartialCloneTo(AlignedTVList cloneList, Set<Integer> retainedColumns) { + copyAdditionalStateTo(cloneList); + } + protected void onColumnsReleased(Set<Integer> retainedColumns) { // Default AlignedTVList implementations sort only by time and do not depend on value columns. } + /** Encoded columns expose logical values through getters without allocating value arrays. */ + protected boolean isEncodedColumn(int columnIndex) { + return false; + } + + protected boolean isValueBlockNull(int columnIndex, int arrayIndex) { + return values.get(columnIndex).get(arrayIndex) == null; + } + private List<Object> convertToType(TSDataType to, TSDataType from, List<Object> originalValues) { if (!to.isCompatible(from)) { return null; @@ -238,6 +255,11 @@ public abstract class AlignedTVList extends TVList { .collect(Collectors.toList()); } + protected AlignedTVList createColumnProjection( + List<Integer> columnIndices, List<TSDataType> types) { + return AlignedTVListFactory.createAlignedList(new ArrayList<>(types)); + } + @Override @TestOnly public TVList getTvListByColumnIndex( @@ -268,8 +290,7 @@ public abstract class AlignedTVList extends TVList { } } } - AlignedTVList alignedTvList = - AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypeList)); + AlignedTVList alignedTvList = createColumnProjection(columnIndexList, dataTypeList); alignedTvList.timestamps = this.timestamps; alignedTvList.indices = this.indices; alignedTvList.values = values; @@ -296,7 +317,7 @@ public abstract class AlignedTVList extends TVList { public synchronized AlignedTVList cloneForFlushSort() { AlignedTVList cloneList = createNewList(new ArrayList<>(dataTypes), true); cloneAs(cloneList); - copyAdditionalStateTo(cloneList); + copyStateForFlushSortTo(cloneList); cloneList.timeDeletedCnt = this.timeDeletedCnt; cloneList.memoryBinaryChunkSize = this.memoryBinaryChunkSize; cloneList.values = this.values; @@ -340,7 +361,7 @@ public abstract class AlignedTVList extends TVList { } } cloneAs(cloneList); - copyAdditionalStateTo(cloneList); + copyStateForPartialCloneTo(cloneList, retainedColumns); cloneColumnDataTo(cloneList, retainedColumns); return prepareMovePlan(cloneList, retainedColumns); } @@ -514,6 +535,9 @@ public abstract class AlignedTVList extends TVList { timestamps.get(arrayIndex)[elementIndex] = timestamp; for (int i = 0; i < values.size(); i++) { Object columnValue = value[i]; + if (isEncodedColumn(i)) { + continue; + } if (columnValue == null) { markNullValue(i, arrayIndex, elementIndex); continue; @@ -616,7 +640,7 @@ public abstract class AlignedTVList extends TVList { case BLOB: case STRING: case OBJECT: - Binary valueT = ((Binary[]) columnValues.get(arrayIndex))[elementIndex]; + Binary valueT = getBinaryByValueIndex(validValueIndex, columnIndex); vector[columnIndex] = TsPrimitiveType.getByType(TSDataType.TEXT, valueT); break; case FLOAT: @@ -730,7 +754,7 @@ public abstract class AlignedTVList extends TVList { case BLOB: case TEXT: case OBJECT: - return ((Binary[]) columnValues.get(arrayIndex))[elementIndex]; + return getBinaryByValueIndex(rowIndex, columnIndex); default: throw new IllegalArgumentException( dataTypes.get(columnIndex) + StorageEngineMessages.IS_NOT_SUPPORTED); @@ -834,7 +858,8 @@ public abstract class AlignedTVList extends TVList { || dataTypes.get(columnIndex) == TSDataType.BLOB || dataTypes.get(columnIndex) == TSDataType.STRING || dataTypes.get(columnIndex) == TSDataType.OBJECT) { - return ((Binary[]) columnValues.get(arrayIndex))[elementIndex]; + Object valueArray = columnValues == null ? null : columnValues.get(arrayIndex); + return valueArray == null ? null : ((Binary[]) valueArray)[elementIndex]; } else { return (Binary) TSDataType.TEXT.castFromSingleValue( @@ -881,7 +906,7 @@ public abstract class AlignedTVList extends TVList { return true; } int arrayIndex = unsortedRowIndex / ARRAY_SIZE; - if (values.get(columnIndex).get(arrayIndex) == null) { + if (isValueBlockNull(columnIndex, arrayIndex)) { return true; } if (bitMaps == null @@ -1074,8 +1099,9 @@ public abstract class AlignedTVList extends TVList { return; } for (int arrayIndex = 0; arrayIndex < columnValues.size(); arrayIndex++) { - if (columnValues.get(arrayIndex) != null) { - getBitMap(columnIndex, arrayIndex).markAll(); + if (!isValueBlockNull(columnIndex, arrayIndex)) { + getBitMap(columnIndex, arrayIndex) + .markRange(0, Math.min(ARRAY_SIZE, rowCount - arrayIndex * ARRAY_SIZE)); } } } @@ -1356,7 +1382,7 @@ public abstract class AlignedTVList extends TVList { for (int j = 0; j < inputValues.length; j++) { // A null value array represents an entirely null block, so no bitmap is needed until the // block receives its first non-null value. - if (values.get(j).get(arrayIndex) == null) { + if (isValueBlockNull(j, arrayIndex)) { continue; } @@ -1421,7 +1447,9 @@ public abstract class AlignedTVList extends TVList { int elementIndex, int remaining) { for (int i = 0; i < values.size(); i++) { - if (value[i] == null || !containsNonNullValue(bitMaps, results, i, idx, remaining)) { + if (isEncodedColumn(i) + || value[i] == null + || !containsNonNullValue(bitMaps, results, i, idx, remaining)) { continue; } Object valueArray = getOrCreateValueArray(i, arrayIndex, elementIndex); @@ -1543,7 +1571,11 @@ public abstract class AlignedTVList extends TVList { private boolean markNullValue(int columnIndex, int arrayIndex, int elementIndex) { List<Object> columnValues = values.get(columnIndex); - if (columnValues == null || columnValues.get(arrayIndex) == null) { + if (columnValues == null || isValueBlockNull(columnIndex, arrayIndex)) { + return false; + } + if (isEncodedColumn(columnIndex) + && isNullValue(arrayIndex * ARRAY_SIZE + elementIndex, columnIndex)) { return false; } // mark the null value in the current bitmap @@ -2056,12 +2088,7 @@ public abstract class AlignedTVList extends TVList { case STRING: case OBJECT: for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) { - int arrayIndex = rowIdx / ARRAY_SIZE; - Object valueArray = values.get(columnIndex).get(arrayIndex); - Binary value = - valueArray == null - ? Binary.EMPTY_VALUE - : ((Binary[]) valueArray)[rowIdx % ARRAY_SIZE]; + Binary value = getBinaryByValueIndex(rowIdx, columnIndex); size += ReadWriteIOUtils.sizeToWrite(value == null ? Binary.EMPTY_VALUE : value); } break; @@ -2135,8 +2162,7 @@ public abstract class AlignedTVList extends TVList { case BLOB: case STRING: case OBJECT: - Binary valueT = - valueArray == null ? Binary.EMPTY_VALUE : ((Binary[]) valueArray)[elementIndex]; + Binary valueT = getBinaryByValueIndex(valueIndex, columnIndex); // In some scenario, the Binary in AlignedTVList will be null if this field is empty in // current row. We need to handle this scenario to get rid of NPE. See the similar issue // here: https://github.com/apache/iotdb/pull/9884 diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java index a146c867fa0..7c0e5f03bae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java @@ -59,9 +59,18 @@ public final class AlignedTVListFactory { return new ColumnSortedAlignedTVList(dataTypes, keyColumnIndices, true); } + /** Creates flat-device storage with a single dictionary-code column for the logical device. */ + public static AlignedTVList createFlatDeviceAlignedList( + List<TSDataType> dataTypes, List<Integer> tagIndices) { + return new FlatDeviceAlignedTVList(dataTypes, tagIndices, true); + } + public static AlignedTVList deserialize(DataInputStream stream) throws IOException { byte encodedDataType = stream.readByte(); int dataTypeCount = stream.readInt(); + if (encodedDataType == FlatDeviceAlignedTVList.WAL_DATA_TYPE_MARKER) { + return ColumnSortedAlignedTVList.deserialize(stream, dataTypeCount, true); + } // The dedicated marker is the only discriminator for the column-sorted WAL format. VECTOR // continues to identify the ordinary AlignedTVList format; its count is never overloaded. if (encodedDataType == ColumnSortedAlignedTVList.WAL_DATA_TYPE_MARKER) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java index 2d0e7da1c27..50f762929a4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java @@ -39,7 +39,7 @@ import java.util.Set; import java.util.function.IntPredicate; /** - * An aligned TVList that orders rows by timestamp followed by selected value columns. Selected + * An aligned TVList that orders rows by selected value columns followed by timestamp. Selected * columns are compared in the configured order, with null values before non-null values. */ public class ColumnSortedAlignedTVList extends AlignedTVList { @@ -50,8 +50,7 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { private final List<Integer> sortColumnIndices; private final TimSort policy; - // Keep value-column ordering separate so timestamp-only binary searches retain their prefix - // guarantees even when equal-time rows need a compound-key sort. + // The compound-key prefix is independent of timestamp order: time can decrease between TAGs. private boolean valueColumnsSorted = true; /** @@ -143,7 +142,7 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { @Override public synchronized boolean isSorted() { - return sorted && valueColumnsSorted; + return valueColumnsSorted; } @Override @@ -154,7 +153,9 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { seqRowCount = previousSeqRowCount; } else if (rowCount > 1 && compareRows(rowCount - 2, rowCount - 1) > 0) { valueColumnsSorted = false; - seqRowCount = Math.min(seqRowCount, rowCount - 1); + seqRowCount = rowCount - 1; + } else { + seqRowCount = rowCount; } } @@ -171,10 +172,11 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { for (int i = Math.max(1, previousRowCount); i < rowCount; i++) { if (compareRows(i - 1, i) > 0) { valueColumnsSorted = false; - seqRowCount = Math.min(seqRowCount, i); + seqRowCount = i; return; } } + seqRowCount = rowCount; } @Override @@ -254,6 +256,11 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { @SuppressWarnings("java:S6541") static ColumnSortedAlignedTVList deserialize(DataInputStream stream, int dataTypeCount) throws IOException { + return deserialize(stream, dataTypeCount, false); + } + + static ColumnSortedAlignedTVList deserialize( + DataInputStream stream, int dataTypeCount, boolean flatDevice) throws IOException { List<TSDataType> dataTypes = new ArrayList<>(dataTypeCount); for (int columnIndex = 0; columnIndex < dataTypeCount; columnIndex++) { dataTypes.add(ReadWriteIOUtils.readDataType(stream)); @@ -264,7 +271,9 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { sortColumnIndices.add(stream.readInt()); } ColumnSortedAlignedTVList tvList = - new ColumnSortedAlignedTVList(dataTypes, sortColumnIndices, true); + flatDevice + ? new FlatDeviceAlignedTVList(dataTypes, sortColumnIndices, true) + : new ColumnSortedAlignedTVList(dataTypes, sortColumnIndices, true); int rowCount = stream.readInt(); long[] times = new long[rowCount]; @@ -358,7 +367,8 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { } policy.clearSortedValue(); policy.clearSortedTime(); - sorted = true; + // The inherited flag describes timestamps only, not compound-key sortedness. + sorted = sortColumnIndices.isEmpty(); valueColumnsSorted = true; seqRowCount = rowCount; return rowCount; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MemPointIteratorFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MemPointIteratorFactory.java index 71402b7dbe7..43714149986 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MemPointIteratorFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MemPointIteratorFactory.java @@ -619,6 +619,10 @@ public class MemPointIteratorFactory { long time = Long.MIN_VALUE; for (int i = 0; i < tvLists.size(); i++) { TVList list = tvLists.get(i); + // Time endpoints cannot prove that TAG-first lists have disjoint logical-key ranges. + if (list instanceof AlignedTVList alignedList && !alignedList.isTimeOrdered()) { + return false; + } int rowCount = tvListRowCounts == null ? list.rowCount() : tvListRowCounts.get(i); if (rowCount == 0) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortMultiAlignedTVListIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortMultiAlignedTVListIterator.java index f0d31a098e4..0373c6d0ede 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortMultiAlignedTVListIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MergeSortMultiAlignedTVListIterator.java @@ -40,8 +40,6 @@ import java.util.Set; import java.util.stream.Collectors; import java.util.stream.IntStream; -import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted; - public class MergeSortMultiAlignedTVListIterator extends MultiAlignedTVListIterator { private final Set<Integer> probeIterators; private final int[] iteratorIndices; @@ -109,6 +107,12 @@ public class MergeSortMultiAlignedTVListIterator extends MultiAlignedTVListItera alignedTvListIterators.get(rightIteratorIndex); AlignedTVList leftList = leftIterator.getAlignedTVList(); AlignedTVList rightList = rightIterator.getAlignedTVList(); + if (leftIterator.isTimeOrderedScan() && rightIterator.isTimeOrderedScan()) { + int timeComparison = Long.compare(leftIterator.currentTime(), rightIterator.currentTime()); + if (timeComparison != 0) { + return timeComparison; + } + } return leftList.compareRowKey( leftIterator.getScanOrderIndex(leftIterator.getIndex()), rightList, @@ -157,7 +161,7 @@ public class MergeSortMultiAlignedTVListIterator extends MultiAlignedTVListItera } // check valueColumnsDeletionList if (valueColumnsDeletionList != null - && isPointDeleted( + && isDeletedInScanOrder( currentTime, valueColumnsDeletionList.get(columnIndex), valueColumnDeleteCursor.get(columnIndex), @@ -197,7 +201,7 @@ public class MergeSortMultiAlignedTVListIterator extends MultiAlignedTVListItera // check valueColumnsDeletionList if (valueColumnsDeletionList != null - && isPointDeleted( + && isDeletedInScanOrder( currentTime, valueColumnsDeletionList.get(columnIndex), valueColumnDeleteCursor.get(columnIndex), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MultiAlignedTVListIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MultiAlignedTVListIterator.java index 83bfa8c6888..9c3379edcca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MultiAlignedTVListIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/MultiAlignedTVListIterator.java @@ -42,6 +42,8 @@ import java.util.ArrayList; import java.util.List; import java.util.function.LongConsumer; +import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted; + public abstract class MultiAlignedTVListIterator extends MemPointIterator { protected List<TSDataType> tsDataTypeList; protected List<Integer> columnIndexList; @@ -59,6 +61,7 @@ public abstract class MultiAlignedTVListIterator extends MemPointIterator { // used by nextBatch during query protected final int maxNumberOfPointsInPage; protected final List<int[]> valueColumnDeleteCursor; + private final boolean timeOrderedScan; protected MultiAlignedTVListIterator( List<TSDataType> tsDataTypeList, @@ -117,6 +120,9 @@ public abstract class MultiAlignedTVListIterator extends MemPointIterator { alignedTvListIterators.add(iterator); } } + this.timeOrderedScan = + alignedTvListIterators.stream() + .allMatch(AlignedTVList.AlignedTVListIterator::isTimeOrderedScan); this.valueColumnsDeletionList = valueColumnsDeletionList; this.floatPrecision = floatPrecision != null ? floatPrecision : 0; this.encodingList = encodingList; @@ -145,6 +151,18 @@ public abstract class MultiAlignedTVListIterator extends MemPointIterator { return hasNext && !isCurrentTimeExceedTimeRange(currentTime); } + @Override + protected boolean isCurrentTimeExceedTimeRange(long time) { + return timeOrderedScan && super.isCurrentTimeExceedTimeRange(time); + } + + protected boolean isDeletedInScanOrder( + long time, List<TimeRange> deletions, int[] cursor, Ordering ordering) { + return timeOrderedScan + ? isPointDeleted(time, deletions, cursor, ordering) + : isPointDeleted(time, deletions); + } + @Override public TimeValuePair nextTimeValuePair() { if (!hasNextTimeValuePair()) { @@ -306,8 +324,7 @@ public abstract class MultiAlignedTVListIterator extends MemPointIterator { @Override public long getUsedMemorySize() { - // not used - return 0; + return alignedTvListIterators.stream().mapToLong(MemPointIterator::getUsedMemorySize).sum(); } protected abstract int currentIteratorIndex(int columnIndex); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java index 31fde52a9de..eecd5a5ea58 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java @@ -22,7 +22,9 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.tablemodel; import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchema; import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionTestFileWriter; +import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunk; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.ColumnCategory; @@ -41,6 +43,8 @@ import org.apache.tsfile.write.schema.Schema; import java.io.IOException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; public class CompactionTableModelTestFileWriter extends CompactionTestFileWriter { private Schema schema; @@ -142,6 +146,20 @@ public class CompactionTableModelTestFileWriter extends CompactionTestFileWriter chunkWriter.writeToFileWriter(fileWriter); } + public void flushFlatDeviceMemChunk(AlignedWritableMemChunk chunk, int maxPointsInChunk) + throws IOException { + // Use the production flush encoder so page boundaries and statistics reflect TAG-first rows. + currentDeviceStartTime = Math.min(currentDeviceStartTime, chunk.getFirstPoint()); + currentDeviceEndTime = Math.max(currentDeviceEndTime, chunk.getLastPoint()); + chunk.sortTvListForFlush(); + BlockingQueue<Object> encoded = new LinkedBlockingQueue<>(); + chunk.encode( + encoded, new BatchEncodeInfo(0, 0, 0, 100, maxPointsInChunk, 1024 * 1024), new long[100]); + for (Object item : encoded) { + ((AlignedChunkWriterImpl) item).writeToFileWriter(fileWriter); + } + } + public IDeviceID startChunkGroup(String tableName, List<String> idColumnFields) throws IOException { int idx = 0; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java index 2ae57a24fe9..49f13c06e67 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.tablemodel; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.AlignedFullPath; import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil; @@ -35,7 +36,9 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.RepairUnsortedFileCompactionTask; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.reader.CompactionChunkReader; import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.RepairDataFileScanUtil; +import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunk; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileRepairStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -43,13 +46,17 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; +import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.StringArrayDeviceID; import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.block.TsBlock; import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.TsPrimitiveType; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -163,7 +170,7 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { assertFlatDeviceFileSorted(target); Assert.assertEquals( Arrays.asList( - "1:a:20:x", "3:a:21:x", "3:b:11:x", "3:c:22:x", "4:a:12:x", "5:a:13:x", "6:a:23:x"), + "1:a:20:x", "3:a:21:x", "4:a:12:x", "5:a:13:x", "6:a:23:x", "3:b:11:x", "3:c:22:x"), readFlatDeviceRows(target)); IDeviceID tableDevice = new StringArrayDeviceID(new String[] {"flat_table"}); Assert.assertEquals(Collections.singleton(tableDevice), target.getDevices()); @@ -201,6 +208,7 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { for (TsFileResource target : tsFileManager.getTsFileList(true)) { assertFlatDeviceFileSorted(target); rows.addAll(readFlatDeviceRows(target)); + assertFlatDeviceStatistics(target, 3, 3); } Assert.assertEquals( Arrays.asList("1:a:20:x", "1:b:11:y", "1:c:21:z", "3:a:40:x", "3:b:31:y", "3:c:41:z"), @@ -245,10 +253,10 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { ICompactionPerformer performer) throws Exception { // Model overlapping unseq flushes from clients at different time offsets. Each source is - // ordered by (time, tag1, tag2), while equal timestamps cross page/chunk and output-block + // ordered by (TAG, time), while equal timestamps cross page/chunk and output-block // boundaries. Duplicate full keys must retain the value from the newer file. TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(rowsPerPage); - Map<Long, String> expectedRows = new TreeMap<>(); + Map<String, String> expectedRows = new TreeMap<>(); for (int file = 0; file < fileCount; file++) { TsFileResource source = createEmptyFileAndResource(false); unseqResources.add(source); @@ -281,7 +289,8 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { tag2[row] = "sensor_" + device % 4; fields[row] = file * 1_000_000L + key; expectedRows.put( - key, times[row] + ":" + tag1[row] + ":" + fields[row] + ":" + tag2[row]); + tag1[row] + "\u0000" + tag2[row] + "\u0000" + String.format("%020d", times[row]), + times[row] + ":" + tag1[row] + ":" + fields[row] + ":" + tag2[row]); } writer.generateFlatDeviceAlignedRows(times, tag1, fields, tag2, rowsPerPage); } @@ -300,6 +309,7 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { TsFileResource target = tsFileManager.getTsFileList(false).get(0); assertFlatDeviceFileSorted(target); Assert.assertEquals(new ArrayList<>(expectedRows.values()), readFlatDeviceRows(target)); + assertFlatDeviceStatistics(target, deviceCount, expectedRows.size()); IDeviceID tableDevice = new StringArrayDeviceID(new String[] {"flat_table"}); Assert.assertEquals(Collections.singleton(tableDevice), target.getDevices()); Assert.assertEquals(0L, target.getStartTime(tableDevice).get().longValue()); @@ -307,6 +317,28 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { (fileCount - 1) * 11L + timesPerFile - 1, target.getEndTime(tableDevice).get().longValue()); } + private void assertFlatDeviceStatistics(TsFileResource resource, long devices, long rows) + throws IOException { + try (TsFileSequenceReader reader = new TsFileSequenceReader(resource.getTsFilePath())) { + Assert.assertEquals( + devices, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get( + TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey( + "flat_table"))) + .getLong()); + Assert.assertEquals( + rows, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get(TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey("flat_table"))) + .getLong()); + } + } + private void assertFlatDeviceFileSorted(TsFileResource resource) throws IOException { RepairDataFileScanUtil scan = new RepairDataFileScanUtil(resource); scan.scanTsFile(true); @@ -336,7 +368,7 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { RepairDataFileScanUtil scan = new RepairDataFileScanUtil(resource); scan.scanTsFile(checkResource); Assert.assertFalse(scan.isBrokenFile()); - Assert.assertEquals(!"z".equals(secondTag), scan.hasUnsortedDataOrWrongStatistics()); + Assert.assertEquals("m".equals(secondTag), scan.hasUnsortedDataOrWrongStatistics()); } } } @@ -370,7 +402,7 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { Assert.assertEquals(1L, target.getStartTime(tableDevice).get().longValue()); Assert.assertEquals(3L, target.getEndTime(tableDevice).get().longValue()); Assert.assertEquals( - Arrays.asList("1:a:10:x", "1:b:11:y", "2:a:20:x", "3:b:30:y"), readFlatDeviceRows(target)); + Arrays.asList("1:a:10:x", "2:a:20:x", "1:b:11:y", "3:b:30:y"), readFlatDeviceRows(target)); try (TsFileSequenceReader reader = new TsFileSequenceReader(target.getTsFilePath())) { Assert.assertEquals( 2L, @@ -436,6 +468,87 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { Arrays.asList("1:a:10:x", "1:a:20:z", "1:b:21:x", "1:b:11:y"), readFlatDeviceRows(target)); } + // Flush TAG-first rows with regressing time into overlapping pages/chunks. Both forward and + // reverse disk reads, repair validation and INNER_UNSEQ compaction must preserve all row keys. + @Test + public void testTagFirstMemTableFlushAndCompaction() throws Exception { + int oldThreshold = IoTDBDescriptor.getInstance().getConfig().getTvListSortThreshold(); + try { + for (int threshold : new int[] {0, 2}) { + IoTDBDescriptor.getInstance().getConfig().setTVListSortThreshold(threshold); + for (int chunkSize : new int[] {2, 100}) { + TsFileResource resource = createEmptyFileAndResource(threshold == 0); + (resource.isSeq() ? seqResources : unseqResources).add(resource); + List<IMeasurementSchema> schemas = + List.of( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("field", TSDataType.INT64), + new MeasurementSchema("tag2", TSDataType.STRING)); + AlignedWritableMemChunk mem = + new AlignedWritableMemChunk(schemas, true, List.of(0, 2), true); + for (String tag : List.of("b", "a")) { + for (long time : new long[] {100, 1, 50}) { + mem.writeAlignedPoints( + time, + new Object[] { + new Binary(tag, TSFileConfig.STRING_CHARSET), + time, + new Binary("x", TSFileConfig.STRING_CHARSET) + }, + schemas); + } + } + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(resource)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", List.of()); + writer.flushFlatDeviceMemChunk(mem, chunkSize); + writer.endChunkGroup(); + writer.endFile(); + } + if (chunkSize == 100) { + // The six TAG-first rows fit in one page. A TAG switch changes the physical time + // from 100 back to 1, but that regression must not seal an additional page. + Assert.assertEquals(1, countFlatDeviceTimePages(resource)); + } + List<String> expected = + List.of("1:a:1:x", "50:a:50:x", "100:a:100:x", "1:b:1:x", "50:b:50:x", "100:b:100:x"); + Assert.assertEquals(expected, readFlatDeviceRows(resource)); + List<String> descending = new ArrayList<>(expected); + Collections.reverse(descending); + Assert.assertEquals(descending, readFlatDeviceRows(resource, Ordering.DESC)); + assertFlatDeviceFileSorted(resource); + } + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setTVListSortThreshold(oldThreshold); + } + tsFileManager.addAll(unseqResources, false); + Assert.assertTrue( + new InnerSpaceCompactionTask( + 0, tsFileManager, unseqResources, false, new FastCompactionPerformer(false), 0) + .start()); + Assert.assertEquals(1, tsFileManager.getTsFileList(false).size()); + TsFileResource target = tsFileManager.getTsFileList(false).get(0); + assertFlatDeviceFileSorted(target); + Assert.assertEquals( + List.of("1:a:1:x", "50:a:50:x", "100:a:100:x", "1:b:1:x", "50:b:50:x", "100:b:100:x"), + readFlatDeviceRows(target)); + } + + private int countFlatDeviceTimePages(TsFileResource resource) throws IOException { + IDeviceID physicalDevice = new StringArrayDeviceID(new String[] {"flat_table"}); + int pageCount = 0; + try (TsFileSequenceReader reader = new TsFileSequenceReader(resource.getTsFilePath())) { + for (AbstractAlignedChunkMetadata metadata : + reader.getAlignedChunkMetadata(physicalDevice, false)) { + Chunk timeChunk = reader.readMemChunk((ChunkMetadata) metadata.getTimeChunkMetadata()); + pageCount += new CompactionChunkReader(timeChunk).readPageDataWithoutUncompressing().size(); + } + } + return pageCount; + } + private TsFileResource createFlatDeviceFile( boolean sequence, long fieldOffset, String[] tag1, String[] tag2, long timestamp) throws IOException { @@ -453,6 +566,11 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { } private List<String> readFlatDeviceRows(TsFileResource resource) throws IOException { + return readFlatDeviceRows(resource, Ordering.ASC); + } + + private List<String> readFlatDeviceRows(TsFileResource resource, Ordering scanOrder) + throws IOException { IDeviceID physicalDevice = new StringArrayDeviceID(new String[] {"flat_table"}); List<IMeasurementSchema> schemas = Arrays.asList( @@ -466,14 +584,16 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { AlignedSeriesScanUtil scan = new AlignedSeriesScanUtil( path, - Ordering.ASC, + scanOrder, options.build(), FragmentInstanceContext.createFragmentInstanceContextForCompaction(0L), true, Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.INT64), Arrays.asList(0, 1)); scan.initQueryDataSource( - new QueryDataSource(Collections.singletonList(resource), Collections.emptyList())); + new QueryDataSource( + resource.isSeq() ? List.of(resource) : List.of(), + resource.isSeq() ? List.of() : List.of(resource))); List<String> rows = new ArrayList<>(); while (scan.hasNextFile().orElse(true)) { while (scan.hasNextChunk().orElse(true)) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java index 820bc150ec6..556558ee4a1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; @@ -95,7 +96,7 @@ public class FlatDeviceMemTableDeletionTest { assertEquals(2, memTable.delete(deletion)); assertEquals(0, memTable.delete(deletion)); list.sort(); - assertEquals(List.of(5L, 10L, 20L), remainingValues(list, 2)); + assertEquals(List.of(10L, 5L, 20L), remainingValues(list, 2)); assertEquals(0, list.getMinTime()); assertEquals(2, list.getMaxTime()); } @@ -153,7 +154,7 @@ public class FlatDeviceMemTableDeletionTest { insert(memTable, 1, "a", "x", 10); insert(memTable, 1, "b", "x", 20); TableDeletionEntry deletion = - new TableDeletionEntry( + flatDeletion( new DeletionPredicate( TABLE, new TagPredicate.SegmentExactMatch("a", 1), List.of("field1")), new TimeRange(1, 1)); @@ -210,14 +211,15 @@ public class FlatDeviceMemTableDeletionTest { assertEquals( 1, flat.delete( - deletion( + physicalDeletion( new TagPredicate.FullExactMatch( IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {TABLE})), 1, 1))); - assertEquals(1, flat.delete(deletion(new TagPredicate.NOP(), 2, 2))); + assertEquals(1, flat.delete(physicalDeletion(new TagPredicate.NOP(), 2, 2))); assertTrue(flat.getMemTableMap().isEmpty()); - assertEquals(1, ordinary.delete(deletion(new TagPredicate.SegmentExactMatch("a", 1), 1, 1))); + assertEquals( + 1, ordinary.delete(physicalDeletion(new TagPredicate.SegmentExactMatch("a", 1), 1, 1))); assertTrue(ordinary.getMemTableMap().isEmpty()); } @@ -248,10 +250,11 @@ public class FlatDeviceMemTableDeletionTest { try { AlignedTVList list = recovered.getWorkingTVList(); assertTrue(list.isTimeDeleted(0)); - assertEquals(List.of(20L, 30L), remainingValues(list, 1)); + assertEquals(flat ? List.of(30L, 20L) : List.of(20L, 30L), remainingValues(list, 1)); assertEquals(2, list.getMinTime()); assertEquals(3, list.getMaxTime()); - assertEquals(binary("c"), list.getBinaryByValueIndex(list.getValueIndex(1), 0)); + assertEquals( + binary(flat ? "b" : "c"), list.getBinaryByValueIndex(list.getValueIndex(1), 0)); assertEquals(2, recovered.deleteTime(2, 3)); assertTrue(recovered.isAllDeleted()); } finally { @@ -275,7 +278,7 @@ public class FlatDeviceMemTableDeletionTest { assertEquals( 1, memTable.delete( - new TableDeletionEntry( + flatDeletion( new DeletionPredicate( TABLE, new TagPredicate.SegmentExactMatch("b", 1), List.of("tag1")), new TimeRange(1, 1)))); @@ -322,6 +325,18 @@ public class FlatDeviceMemTableDeletionTest { } private static TableDeletionEntry deletion(TagPredicate predicate, long start, long end) { + return flatDeletion(new DeletionPredicate(TABLE, predicate), new TimeRange(start, end)); + } + + private static FlatDeviceDeletionEntry flatDeletion( + DeletionPredicate predicate, TimeRange timeRange) { + // Match AnalyzeUtils: flat predicates carry logical TAG names and are evaluated per row. + FlatDeviceDeletionEntry deletion = new FlatDeviceDeletionEntry(predicate, timeRange); + deletion.setTagColumnNames(List.of("tag1", "tag2")); + return deletion; + } + + private static TableDeletionEntry physicalDeletion(TagPredicate predicate, long start, long end) { return new TableDeletionEntry( new DeletionPredicate(TABLE, predicate), new TimeRange(start, end)); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java index 008da09f2a6..23a9c240b65 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java @@ -22,26 +22,40 @@ package org.apache.iotdb.db.storageengine.dataregion.memtable; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo; import org.apache.iotdb.db.utils.datastructure.ColumnSortedAlignedTVList; +import org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; +import org.apache.tsfile.file.metadata.ChunkMetadata; +import org.apache.tsfile.file.metadata.IChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.BatchData; +import org.apache.tsfile.read.common.Chunk; +import org.apache.tsfile.read.reader.chunk.AlignedChunkReader; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import org.junit.Assert; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import java.io.ByteArrayInputStream; import java.io.DataInputStream; +import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; @@ -52,6 +66,8 @@ import java.util.stream.Collectors; public class FlatDeviceMemTableTest { + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + private static final String TABLE_NAME = "table1"; private static final String[] MEASUREMENTS = {"tag1", "tag2", "attr1", "field1"}; private static final TSDataType[] DATA_TYPES = { @@ -71,7 +87,7 @@ public class FlatDeviceMemTableTest { }; // Covers flat row writes with equal timestamps and different tags. All rows must share the - // table-name group, ATTRIBUTE must be excluded, and sorting must use time then both TAG columns. + // table-name group, ATTRIBUTE must be excluded, and sorting must use both TAG columns then time. @Test public void testFlatRowsUseOneColumnSortedGroup() { PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); @@ -388,15 +404,15 @@ public class FlatDeviceMemTableTest { 11L, chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(0), 1)); Assert.assertEquals( - binary("z"), + binary("x"), chunk .getWorkingTVList() .getBinaryByValueIndex(chunk.getWorkingTVList().getValueIndex(1), 0)); Assert.assertEquals( - 10L, + 20L, chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(1), 1)); Assert.assertEquals( - 20L, + 10L, chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(2), 1)); } @@ -444,6 +460,78 @@ public class FlatDeviceMemTableTest { Assert.assertEquals(2, writer.getTimeChunkWriter().getStatistics().getCount()); } + @Test + public void testDictionaryFlushExpandsTagChunksInBothEncodePaths() throws Exception { + int oldThreshold = IoTDBDescriptor.getInstance().getConfig().getTvListSortThreshold(); + try { + for (int threshold : new int[] {0, 2}) { + IoTDBDescriptor.getInstance().getConfig().setTVListSortThreshold(threshold); + List<IMeasurementSchema> schemas = + List.of( + new MeasurementSchema("a_tag", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("b_tag", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("c_field", TSDataType.INT64)); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true, 2, true); + chunk.writeAlignedPoints(1, new Object[] {binary("b"), binary("x"), 10L}, schemas); + chunk.writeAlignedPoints(9, new Object[] {binary("a"), binary("y"), 90L}, schemas); + chunk.writeAlignedPoints(1, new Object[] {binary("b"), binary("x"), 11L}, schemas); + chunk.writeAlignedPoints(1, new Object[] {binary("b"), binary("x"), null}, schemas); + chunk.writeAlignedPoints(2, new Object[] {binary("b"), null, 20L}, schemas); + Assert.assertTrue(chunk.getWorkingTVList() instanceof FlatDeviceAlignedTVList); + if (threshold > 0) { + Assert.assertFalse(chunk.getSortedList().isEmpty()); + } + chunk.sortTvListForFlush(); + BlockingQueue<Object> queue = new LinkedBlockingQueue<>(); + chunk.encode(queue, new BatchEncodeInfo(0, 0, 0, 100, 100, 1024 * 1024), new long[100]); + File file = temporaryFolder.newFile("dictionary-" + threshold + ".tsfile"); + IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.flat"); + try (TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer.startChunkGroup(device); + for (Object task : queue) { + ((AlignedChunkWriterImpl) task).writeToFileWriter(writer); + } + writer.endChunkGroup(); + writer.endFile(); + } + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getPath())) { + int row = 0; + long[] expectedTimes = {9, 2, 1}; + long[] expectedFields = {90, 20, 11}; + String[] firstTags = {"a", "b", "b"}; + String[] secondTags = {"y", null, "x"}; + for (AbstractAlignedChunkMetadata metadata : + reader.getAlignedChunkMetadata(device, true)) { + Chunk time = reader.readMemChunk((ChunkMetadata) metadata.getTimeChunkMetadata()); + List<Chunk> values = new ArrayList<>(); + for (IChunkMetadata value : metadata.getValueChunkMetadataList()) { + values.add(reader.readMemChunk((ChunkMetadata) value)); + } + AlignedChunkReader chunkReader = new AlignedChunkReader(time, values); + while (chunkReader.hasNextSatisfiedPage()) { + BatchData batch = chunkReader.nextPageData(); + while (batch.hasCurrent()) { + Assert.assertEquals(expectedTimes[row], batch.currentTime()); + Assert.assertEquals(binary(firstTags[row]), batch.getVector()[0].getBinary()); + if (secondTags[row] == null) { + Assert.assertNull(batch.getVector()[1]); + } else { + Assert.assertEquals(binary(secondTags[row]), batch.getVector()[1].getBinary()); + } + Assert.assertEquals(expectedFields[row], batch.getVector()[2].getLong()); + row++; + batch.next(); + } + } + } + Assert.assertEquals(3, row); + } + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setTVListSortThreshold(oldThreshold); + } + } + private static RelationalInsertRowNode newRow(Binary tag1, Binary tag2, long fieldValue) { RelationalInsertRowNode node = new RelationalInsertRowNode( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java index 4e553186a08..d642ef4ac1c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java @@ -27,7 +27,9 @@ import org.apache.tsfile.common.conf.TSFileConfig; 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.common.TimeRange; import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.filter.factory.TimeFilterApi; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -49,10 +51,9 @@ public class ColumnSortedAlignedTVListTest { private static final List<TSEncoding> COMPOUND_ROW_ENCODINGS = List.of(TSEncoding.PLAIN, TSEncoding.PLAIN); - // Covers timestamp-first, multi-column, null-first ordering with payload-bearing rows; equal - // timestamps must follow both configured keys while different timestamps remain time ordered. + // Covers TAG-first, multi-column, null-first ordering; time may decrease between TAG tuples. @Test - public void testSortByTimeAndSelectedColumns() { + public void testSortBySelectedColumnsAndTime() { AlignedTVList tvList = AlignedTVListFactory.createColumnSortedAlignedList( List.of(TSDataType.INT32, TSDataType.TEXT, TSDataType.INT64), 2); @@ -64,8 +65,8 @@ public class ColumnSortedAlignedTVListTest { tvList.sort(); - Assert.assertArrayEquals(new long[] {1, 1, 1, 1, 2}, times(tvList)); - Assert.assertArrayEquals(new long[] {9, 10, 11, 12, 20}, payloads(tvList)); + Assert.assertArrayEquals(new long[] {1, 2, 1, 1, 1}, times(tvList)); + Assert.assertArrayEquals(new long[] {9, 20, 10, 11, 12}, payloads(tvList)); } // Covers every scalar storage representation accepted as a sort key; each larger value is @@ -389,6 +390,187 @@ public class ColumnSortedAlignedTVListTest { Assert.assertEquals(List.of(0, 2), recovered.getSortColumnIndices()); } + // A decreasing timestamp at an increasing TAG boundary is already compound-key ordered. + // Appending a smaller TAG at a later time must instead invalidate the sorted prefix. + @Test + public void testSortedPrefixTracksTagsBeforeTimeForRowsAndTablets() { + AlignedTVList list = newCompoundList(); + list.putAlignedValue(100, new Object[] {binary("a"), 1L}); + list.putAlignedValue(1, new Object[] {binary("b"), 2L}); + Assert.assertTrue(list.isSorted()); + Assert.assertEquals(2, list.seqRowCount()); + list.putAlignedValues( + new long[] {2, 200}, + new Object[] {new Binary[] {binary("b"), binary("a")}, new long[] {3, 4}}, + null, + 0, + 2, + null); + Assert.assertFalse(list.isSorted()); + Assert.assertEquals(3, list.seqRowCount()); + list.sort(); + Assert.assertArrayEquals(new long[] {100, 200, 1, 2}, times(list)); + Assert.assertEquals(1, list.getMinTime()); + Assert.assertEquals(200, list.getMaxTime()); + } + + // WAL snapshots after TAG sorting must keep values paired with regressing timestamps, and + // recovery must rebuild the compound sorted prefix rather than a timestamp-only prefix. + @Test + public void testTagFirstWalRoundTrip() throws IOException { + AlignedTVList list = newCompoundList(); + list.putAlignedValue(1, new Object[] {binary("b"), 1L}); + list.putAlignedValue(100, new Object[] {binary("a"), 100L}); + list.sort(); + ByteBuffer buffer = ByteBuffer.allocate(list.serializedSize()); + list.serializeToWAL(new WALByteBufferForTest(buffer)); + AlignedTVList recovered = deserialize(buffer.array()); + Assert.assertTrue(recovered.isSorted()); + Assert.assertArrayEquals(new long[] {100, 1}, times(recovered)); + Assert.assertArrayEquals(new long[] {100, 1}, payloads(recovered)); + } + + // Raw TAG-ordered scans must not stop at a high timestamp or advance deletion cursors past + // matching rows in the following TAG. Exercise both scan directions and a page time range. + @Test + public void testTagOrderedRangeAndDeletionAcrossTimeRegression() throws IOException { + AlignedTVList list = newCompoundList(); + for (String tag : List.of("a", "b")) { + for (long time : new long[] {1, 2, 100}) { + list.putAlignedValue(time, new Object[] {binary(tag), time}); + } + } + list.sort(); + for (Ordering order : Ordering.values()) { + MemPointIterator iterator = + list.iterator( + order, + list.rowCount(), + TimeFilterApi.gtEq(1), + COMPOUND_ROW_TYPES, + null, + List.of(new TimeRange(1, 1)), + List.of(List.of(), List.of(new TimeRange(2, 2))), + 0, + COMPOUND_ROW_ENCODINGS, + false, + 2, + null); + iterator.setCurrentPageTimeRange(new TimeRange(1, 2)); + List<String> tags = new ArrayList<>(); + while (iterator.hasNextTimeValuePair()) { + TimeValuePair point = iterator.nextTimeValuePair(); + Assert.assertEquals(2, point.getTimestamp()); + Assert.assertNull(point.getValue().getVector()[1]); + tags.add( + point + .getValue() + .getVector()[0] + .getBinary() + .getStringValue(TSFileConfig.STRING_CHARSET)); + } + Assert.assertEquals(order.isAscending() ? List.of("a", "b") : List.of("b", "a"), tags); + } + } + + // TAG-first queries preserve sparse last-write-wins merging across lists and filter every TAG + // within the requested time range, in both scan directions, without a time-sorted permutation. + @Test + public void testTagFirstQueryMergesAndFiltersTimeRangesAcrossLists() throws IOException { + AlignedTVList older = newCompoundList(); + older.putAlignedValue(100, new Object[] {binary("a"), 100L}); + older.putAlignedValue(1, new Object[] {binary("b"), 1L}); + older.putAlignedValue(50, new Object[] {binary("b"), 50L}); + older.sort(); + AlignedTVList newer = newCompoundList(); + newer.putAlignedValue(50, new Object[] {binary("b"), 51L}); + newer.putAlignedValue(50, new Object[] {binary("b"), null}); + newer.putAlignedValue(2, new Object[] {binary("c"), 2L}); + newer.sort(); + for (Ordering order : Ordering.values()) { + MemPointIterator iterator = + MemPointIteratorFactory.create( + COMPOUND_ROW_TYPES, + null, + List.of(older, newer), + null, + order, + null, + null, + null, + 0, + COMPOUND_ROW_ENCODINGS, + false, + 2, + new QueryContext(false, false)); + iterator.setCurrentPageTimeRange(new TimeRange(2, 50)); + List<Long> actual = new ArrayList<>(); + while (iterator.hasNextBatch()) { + TsBlock block = iterator.nextBatch(); + for (int i = 0; i < block.getPositionCount(); i++) { + actual.add(block.getColumn(1).getLong(i)); + } + } + Assert.assertEquals(order.isAscending() ? List.of(51L, 2L) : List.of(2L, 51L), actual); + } + Assert.assertArrayEquals(new long[] {100, 1, 50}, times(older)); + } + + // Timestamp-disjoint TVLists can still have interleaving TAG keys; concatenating them would + // break the TAG-first flush order even though their first/last times appear disjoint. + @Test + public void testFlushMergeDoesNotConcatenateByTimeBounds() throws IOException { + AlignedTVList older = newCompoundList(); + older.putAlignedValue(1, new Object[] {binary("z"), 1L}); + AlignedTVList newer = newCompoundList(); + newer.putAlignedValue(100, new Object[] {binary("a"), 100L}); + older.sort(); + newer.sort(); + MemPointIterator iterator = + MemPointIteratorFactory.create( + COMPOUND_ROW_TYPES, null, List.of(older, newer), false, 10, null); + Assert.assertEquals(100, iterator.nextTimeValuePair().getTimestamp()); + Assert.assertEquals(1, iterator.nextTimeValuePair().getTimestamp()); + Assert.assertFalse(iterator.hasNextTimeValuePair()); + } + + // A single-list query can project away every TAG. Its row permutation must still preserve the + // full logical key and seek correctly in both directions while the physical times regress. + @Test + public void testSingleListProjectedQueryBatchesUseTimePermutation() { + AlignedTVList list = newCompoundList(); + list.putAlignedValue(100, new Object[] {binary("a"), 100L}); + list.putAlignedValue(1, new Object[] {binary("b"), 1L}); + list.putAlignedValue(50, new Object[] {binary("b"), 50L}); + list.sort(); + for (Ordering order : Ordering.values()) { + MemPointIterator iterator = + list.iterator( + order, + list.rowCount(), + null, + List.of(TSDataType.INT64), + List.of(1), + null, + null, + 0, + List.of(TSEncoding.PLAIN), + false, + 1, + new QueryContext(false, false)); + iterator.setCurrentPageTimeRange(new TimeRange(1, 50)); + List<Long> values = new ArrayList<>(); + while (iterator.hasNextBatch()) { + TsBlock batch = iterator.nextBatch(); + for (int i = 0; i < batch.getPositionCount(); i++) { + values.add(batch.getColumn(0).getLong(i)); + } + } + Assert.assertEquals(order.isAscending() ? List.of(1L, 50L) : List.of(50L, 1L), values); + } + Assert.assertArrayEquals(new long[] {100, 1, 50}, times(list)); + } + private static void assertTypeOrdering(TSDataType dataType, Object larger, Object smaller) { AlignedTVList tvList = AlignedTVListFactory.createColumnSortedAlignedList(List.of(dataType, TSDataType.INT64), 1);
