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 38f915ab0ff720b1510ac5f3740951904d369e14 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 22 19:06:56 2026 +0800 tmp --- .../execution/operator/source/SeriesScanUtil.java | 170 ++++++++++++++++++++- .../distribute/TableDistributedPlanGenerator.java | 5 + .../writer/RepairUnsortedFileCompactionWriter.java | 2 +- .../compaction/repair/RepairDataFileScanUtil.java | 31 +++- .../memtable/AlignedWritableMemChunk.java | 61 ++++++-- .../memtable/AlignedWritableMemChunkGroup.java | 54 ++----- .../dataregion/memtable/TsFileProcessor.java | 19 ++- .../utils/datastructure/AlignedTVListFactory.java | 15 +- .../datastructure/ColumnSortedAlignedTVList.java | 11 +- .../tablemodel/FlatDeviceCompactionTest.java | 7 +- .../memtable/FlatDeviceMemTableDeletionTest.java | 14 +- .../memtable/FlatDeviceMemTableTest.java | 19 ++- .../memtable/MemChunkDeserializeTest.java | 3 +- .../modification/TableDeletionEntryTest.java | 48 ++++++ .../tsfile/FlatDeviceAlignedChunkWriterTest.java | 118 ++++++++++++++ 15 files changed, 476 insertions(+), 101 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index 515f6f92814..cf58d2d26ca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -53,6 +53,7 @@ import org.apache.iotdb.db.utils.datastructure.MemPointIterator; import org.apache.tsfile.block.column.Column; 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.IChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.IMetadata; @@ -90,6 +91,8 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.LinkedList; @@ -98,6 +101,7 @@ import java.util.Objects; import java.util.Optional; import java.util.PriorityQueue; import java.util.function.ToLongFunction; +import java.util.stream.Collectors; import static com.google.common.base.Preconditions.checkState; import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.BUILD_TSBLOCK_FROM_MERGE_READER_ALIGNED; @@ -114,6 +118,7 @@ public class SeriesScanUtil implements Accountable { private final IDeviceID deviceID; protected boolean isAligned = false; + protected final boolean currentDeviceIsFlat; protected int[] rowKeyColumnIndices = new int[0]; private final TSDataType dataType; @@ -146,6 +151,7 @@ public class SeriesScanUtil implements Accountable { // result cache private boolean hasCachedNextOverlappedPage; private TsBlock cachedTsBlock; + private FlatDevicePageReaderQueue flatPageReaders; protected SeriesScanOptions scanOptions; private final PaginationController paginationController; @@ -171,6 +177,17 @@ public class SeriesScanUtil implements Accountable { Ordering scanOrder, SeriesScanOptions scanOptions, FragmentInstanceContext context) { + this(seriesPath, scanOrder, scanOptions, context, false); + } + + protected SeriesScanUtil( + IFullPath seriesPath, + Ordering scanOrder, + SeriesScanOptions scanOptions, + FragmentInstanceContext context, + boolean currentDeviceIsFlat) { + // The merge-reader factories are called below, before subclass fields can be initialized. + this.currentDeviceIsFlat = currentDeviceIsFlat; this.seriesPath = seriesPath; this.deviceID = seriesPath.getDeviceId(); this.dataType = seriesPath.getSeriesType(); @@ -294,6 +311,9 @@ public class SeriesScanUtil implements Accountable { // Optional.empty(), it needs to return directly to the checkpoint method that checks the operator // execution time slice. public Optional<Boolean> hasNextFile() throws IOException { + if (rowKeyColumnIndices.length > 0) { + return Optional.of(hasNextFlatDevicePage()); + } if (runtimeFilterExhausted || !paginationController.hasCurLimit()) { return Optional.of(false); } @@ -347,6 +367,9 @@ public class SeriesScanUtil implements Accountable { } public boolean canUseCurrentFileStatistics() { + if (rowKeyColumnIndices.length > 0) { + return false; + } checkState( firstTimeSeriesMetadata != null, DataNodeQueryMessages.EXCEPTION_NO_FIRST_FILE_F5F2E276); @@ -408,6 +431,9 @@ public class SeriesScanUtil implements Accountable { * @throws IllegalStateException illegal state */ public Optional<Boolean> hasNextChunk() throws IOException { + if (rowKeyColumnIndices.length > 0) { + return Optional.of(hasNextFlatDevicePage()); + } if (runtimeFilterExhausted || !paginationController.hasCurLimit()) { return Optional.of(false); } @@ -551,6 +577,9 @@ public class SeriesScanUtil implements Accountable { } public boolean canUseCurrentChunkStatistics() { + if (rowKeyColumnIndices.length > 0) { + return false; + } checkState(firstChunkMetadata != null, DataNodeQueryMessages.EXCEPTION_NO_FIRST_CHUNK_7DCEB14C); if (currentChunkOverlapped() || firstChunkMetadata.isModified()) { @@ -585,6 +614,9 @@ public class SeriesScanUtil implements Accountable { @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning public boolean hasNextPage() throws IOException { + if (rowKeyColumnIndices.length > 0) { + return hasNextFlatDevicePage(); + } if (runtimeFilterExhausted || !paginationController.hasCurLimit()) { return false; } @@ -662,6 +694,14 @@ public class SeriesScanUtil implements Accountable { return false; } + // A flat-device page is physically written in TAG-first order. Its timestamp statistics may + // still describe a non-overlapping range, but returning the page directly would expose that + // physical order to the caller. Route every flat-device page through the row-key merge reader + // so that the page adapter can restore the logical (time, TAG...) order first. + if (rowKeyColumnIndices.length > 0) { + return true; + } + long endpointTime = orderUtils.getOverlapCheckTime(firstPageReader.getStatistics()); unpackAllOverlappedTsFilesToTimeSeriesMetadata(endpointTime); unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(endpointTime, false); @@ -690,7 +730,7 @@ public class SeriesScanUtil implements Accountable { boolean hasMeetSeq = false; while (!cachedChunkMetadata.isEmpty() && orderUtils.isOverlapped(endpointTime, cachedChunkMetadata.peek().getStatistics())) { - if (cachedChunkMetadata.peek().isSeq() && hasMeetSeq) { + if (rowKeyColumnIndices.length == 0 && cachedChunkMetadata.peek().isSeq() && hasMeetSeq) { break; } else if (cachedChunkMetadata.peek().isSeq()) { hasMeetSeq = true; @@ -736,7 +776,23 @@ public class SeriesScanUtil implements Accountable { // init TsBlockBuilder for each page reader pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); - if (chunkMetaData.isSeq()) { + if (rowKeyColumnIndices.length > 0) { + // TAG-first flushes can produce overlapping pages even in a sequence file. Queue them by + // their actual time ranges, but retain the file's sequence flag for version precedence. + for (IPageReader pageReader : pageReaderList) { + unSeqPageReaders.add( + new VersionPageReader( + context, + timestampInFileName, + chunkMetaData.getVersion(), + chunkMetaData.getOffsetOfChunkHeader(), + pageReader, + chunkMetaData.isSeq(), + tsFile, + deletionFilter, + false)); + } + } else if (chunkMetaData.isSeq()) { if (orderUtils.getAscending()) { for (IPageReader iPageReader : pageReaderList) { seqPageReaders.add( @@ -788,6 +844,74 @@ public class SeriesScanUtil implements Accountable { } } + private boolean hasNextFlatDevicePage() throws IOException { + if (!paginationController.hasCurLimit()) { + if (flatPageReaders != null) { + flatPageReaders.clear(); + } + mergeReader.close(); + return false; + } + if (hasCachedNextOverlappedPage) { + return true; + } + if (flatPageReaders == null) { + flatPageReaders = + new FlatDevicePageReaderQueue( + context, + getTsDataTypeList(), + Arrays.stream(rowKeyColumnIndices).boxed().collect(Collectors.toList()), + orderUtils.getAscending(), + scanOptions.getGlobalTimeFilter(), + mergeReader); + // Time indexes remain useful for filtering, but cannot establish TAG order between files, + // chunks or pages. Discover compound page bounds before emitting the first row. + List<TsFileResource> resources = new ArrayList<>(dataSource.getSeqResources()); + resources.addAll(dataSource.getUnseqResources()); + for (TsFileResource resource : resources) { + ITimeSeriesMetadata metadata = loadTimeSeriesMetadata(resource, resource.isSeq()); + if (metadata == null) { + continue; + } + for (IChunkMetadata chunk : FileLoaderUtils.loadChunkMetadataList(metadata)) { + if (chunk.getChunkLoader() instanceof MemChunkLoader memLoader) { + // MemTable iterators already merge by TAG/time. Keep a streaming iterator per mem + // chunk instead of dividing it into overlapping time ranges with a shared cursor. + MemPointIterator points = + memLoader + .getReadOnlyMemChunk() + .createMemPointIterator( + orderUtils.getScanOrder(), scanOptions.getGlobalTimeFilter()); + mergeReader.addReader( + points, + new MergeReaderPriority( + Long.MAX_VALUE, + chunk.getVersion(), + chunk.getOffsetOfChunkHeader(), + chunk.isSeq()), + orderUtils.getOverlapCheckTime(chunk.getStatistics())); + } else if (chunk instanceof AbstractAlignedChunkMetadata alignedChunk) { + flatPageReaders.addChunk(alignedChunk, getFlatDeviceDeletionFilter(resource)); + } + } + } + } + TsBlockBuilder builder = new TsBlockBuilder(getTsDataTypeList()); + while (builder.getPositionCount() < MAX_NUMBER_OF_POINTS_IN_PAGE + && !builder.isFull() + && flatPageReaders.hasNext()) { + context + .getQueryStatistics() + .getPageReaderMaxUsedMemorySize() + .updateAndGet(size -> Math.max(size, mergeReader.getUsedMemorySize())); + addTimeValuePairToResult(mergeReader.nextTimeValuePair(), builder); + } + hasCachedNextOverlappedPage = !builder.isEmpty(); + cachedTsBlock = + hasCachedNextOverlappedPage ? getTransferedDataTypeTsBlock(builder.build()) : null; + return hasCachedNextOverlappedPage; + } + private Filter getFlatDeviceDeletionFilter(TsFileResource resource) { List<FlatDeviceDeletionEntry> deletions; try { @@ -899,6 +1023,9 @@ public class SeriesScanUtil implements Accountable { @SuppressWarnings("unchecked") public boolean canUseCurrentPageStatistics() throws IOException { + if (rowKeyColumnIndices.length > 0) { + return false; + } Statistics<? extends Serializable> currentPageStatistics = currentPageTimeStatistics(); if (currentPageStatistics == null) { return false; @@ -927,6 +1054,11 @@ public class SeriesScanUtil implements Accountable { } public void skipCurrentPage() { + if (rowKeyColumnIndices.length > 0) { + cachedTsBlock = null; + hasCachedNextOverlappedPage = false; + return; + } firstPageReader = null; } @@ -950,6 +1082,11 @@ public class SeriesScanUtil implements Accountable { * `firstPageReader.addPushDownFilter(...)` to ensure the value filter is applied. */ public TsBlock nextPage() throws IOException { + if (rowKeyColumnIndices.length > 0 && !hasCachedNextOverlappedPage) { + if (!hasNextFlatDevicePage()) { + return null; + } + } if (hasCachedNextOverlappedPage) { hasCachedNextOverlappedPage = false; @@ -1029,6 +1166,20 @@ public class SeriesScanUtil implements Accountable { if (filter == null) { return tsBlock; } + if (rowKeyColumnIndices.length > 0) { + // TAG changes can restart time. A rejected row cannot exhaust the rest of this block or scan. + TsBlockBuilder builder = new TsBlockBuilder(getTsDataTypeList()); + for (int row = 0; row < tsBlock.getPositionCount(); row++) { + if (filter.mayQualify(tsBlock.getTimeByIndex(row))) { + builder.getTimeColumnBuilder().writeLong(tsBlock.getTimeByIndex(row)); + for (int column = 0; column < tsBlock.getValueColumnCount(); column++) { + builder.getColumnBuilder(column).write(tsBlock.getColumn(column), row); + } + builder.declarePosition(); + } + } + return builder.build(); + } int positionCount = tsBlock.getPositionCount(); int keepCount = positionCount; @@ -2159,6 +2310,19 @@ public class SeriesScanUtil implements Accountable { boolean isSeq, File tsFile, Filter deletionFilter) { + this(context, fileTimestamp, version, offset, data, isSeq, tsFile, deletionFilter, true); + } + + VersionPageReader( + QueryContext context, + long fileTimestamp, + long version, + long offset, + IPageReader data, + boolean isSeq, + File tsFile, + Filter deletionFilter, + boolean pagesOrderedByTime) { this.context = context; this.version = new MergeReaderPriority(fileTimestamp, version, offset, isSeq); this.data = data; @@ -2167,7 +2331,7 @@ public class SeriesScanUtil implements Accountable { // Overlapping pages must apply file-scoped deletions before entering the merge reader. data.addRecordFilter(deletionFilter); } - this.isSeq = isSeq; + this.isSeq = isSeq && pagesOrderedByTime; this.isAligned = data instanceof AlignedPageReader || data instanceof MemAlignedPageReader diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java index 58e9a48d8ce..f3f241f75c2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java @@ -2143,6 +2143,11 @@ public class TableDistributedPlanGenerator final List<PlanNode> resultTableScanNodeList, final SortPropertyContext sortPropertyContext, final boolean sortDeviceEntries) { + if (deviceTableScanNode.isFlatDevice()) { + // One physical flat device contains many TAG tuples. Its TAG/time scan order does not + // imply global time order, so keep explicit SQL Sort/TopK operators above the scan. + return; + } final Map<Symbol, ColumnSchema> tableColumnSchema; if (deviceTableScanNode instanceof ExternalTsFileScanNode) { tableColumnSchema = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java index e459e2a4f57..3b8e8c279e6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/RepairUnsortedFileCompactionWriter.java @@ -81,7 +81,7 @@ public class RepairUnsortedFileCompactionWriter extends ReadPointInnerCompaction .collect(Collectors.toList()); comparator = (left, right) -> - AlignedPriorityMergeReader.compareAlignedRowKey( + AlignedPriorityMergeReader.compareFlatDeviceRowKey( left, right, types, rowKeyColumnIndices); } dataOfCurrentSeries.sort(comparator); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java index 2b94bb5c42c..fabb79cd615 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.repair; +import java.util.Objects; import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionLastTimeCheckFailedException; @@ -58,6 +59,7 @@ import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.read.reader.chunk.TableChunkReader; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.TsPrimitiveType; @@ -260,16 +262,12 @@ public class RepairDataFileScanUtil { throws IOException { TableSchema schema = reader.getTableSchemaMap().get(device.getTableName()); List<String> tags = new ArrayList<>(); - List<TSDataType> types = new ArrayList<>(); - List<Integer> indices = new ArrayList<>(); for (int i = 0; i < schema.getColumnTypes().size(); i++) { if (schema.getColumnTypes().get(i) == ColumnCategory.TAG) { - // Only TAG chunks are read below, so indices refer to the projected TAG vector. - indices.add(tags.size()); tags.add(schema.getColumnSchemas().get(i).getMeasurementName()); - types.add(schema.getColumnSchemas().get(i).getType()); } } + TimeValuePair previous = null; for (AbstractAlignedChunkMetadata metadata : reader.getAlignedChunkMetadata(device, false)) { Map<String, IChunkMetadata> valueMetadata = new HashMap<>(); @@ -293,8 +291,8 @@ public class RepairDataFileScanUtil { while (rows.hasNextTimeValuePair()) { TimeValuePair row = rows.nextTimeValuePair(); if (previous != null - && AlignedPriorityMergeReader.compareAlignedRowKey(previous, row, types, indices) - >= 0) { + && compareAlignedRowKey(previous, row) + >= 0) { throw new CompactionLastTimeCheckFailedException( device.toString(), row.getTimestamp(), previous.getTimestamp()); } @@ -304,6 +302,19 @@ public class RepairDataFileScanUtil { } } + private int compareAlignedRowKey(TimeValuePair previous, TimeValuePair current) { + TsPrimitiveType[] previousValues = previous.getValue().getVector(); + TsPrimitiveType[] currentValues = current.getValue().getVector(); + for (int i = 0; i < previousValues.length; i++) { + int compare = Objects.compare(previousValues[i].getBinary(), currentValues[i].getBinary(), + Binary::compareTo); + if (compare != 0) { + return compare; + } + } + return Long.compare(previous.getTimestamp(), current.getTimestamp()); + } + private void checkTimeChunkInAlignedSeries( TsFileSequenceReader reader, IDeviceID device, ChunkMetadata timeChunkMetadata) throws IOException { @@ -482,7 +493,11 @@ public class RepairDataFileScanUtil { long currentTime = decoder.readLong(uncompressedTimeData); actualStartTime = Math.min(actualStartTime, currentTime); actualEndTime = Math.max(actualEndTime, currentTime); - checkPreviousTimeAndUpdate(device, currentTime); + // TAG-first flat pages may contain rows from different logical devices, so timestamps can + // go backwards inside one physical page. checkFlatDeviceRows validates each TAG tuple. + if (!currentDeviceIsFlat) { + checkPreviousTimeAndUpdate(device, currentTime); + } } if (actualStartTime > actualEndTime) { return; 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 41d8d2f7e62..f012e9c3561 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 @@ -36,6 +36,7 @@ import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.db.utils.datastructure.AlignedTVListFactory; 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.iotdb.db.utils.datastructure.MemPointIterator; import org.apache.iotdb.db.utils.datastructure.MemPointIteratorFactory; import org.apache.iotdb.db.utils.datastructure.TVList; @@ -114,6 +115,22 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { EncryptUtils.getEncryptParameter()); } + @TestOnly + public AlignedWritableMemChunk( + List<IMeasurementSchema> schemaList, + boolean isTableModel, + int sortColumnCount, + boolean useColumnSortedList, + IDeviceID physicalDeviceId) { + this( + schemaList, + isTableModel, + createContiguousSortColumnIndices(sortColumnCount), + useColumnSortedList, + physicalDeviceId, + EncryptUtils.getEncryptParameter()); + } + @TestOnly public AlignedWritableMemChunk( List<IMeasurementSchema> schemaList, @@ -129,6 +146,22 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { EncryptUtils.getEncryptParameter()); } + @TestOnly + public AlignedWritableMemChunk( + List<IMeasurementSchema> schemaList, + boolean isTableModel, + List<Integer> sortColumnIndices, + boolean useColumnSortedList, + IDeviceID physicalDeviceId) { + this( + schemaList, + isTableModel, + sortColumnIndices, + useColumnSortedList, + physicalDeviceId, + EncryptUtils.getEncryptParameter()); + } + public AlignedWritableMemChunk( List<IMeasurementSchema> schemaList, boolean isTableModel, @@ -178,7 +211,8 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.list = !useColumnSortedList ? AlignedTVListFactory.createAlignedList(dataTypes) - : AlignedTVListFactory.createFlatDeviceAlignedList(dataTypes, this.sortColumnIndices); + : AlignedTVListFactory.createFlatDeviceAlignedList( + dataTypes, this.sortColumnIndices, physicalDeviceId); this.sortedList = new ArrayList<>(); this.ignoreAllNullRows = !isTableModel; this.encryptParameter = encryptParameter; @@ -192,20 +226,17 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { return indices; } - private AlignedWritableMemChunk( - List<IMeasurementSchema> schemaList, AlignedTVList list, boolean isTableModel) { - this(schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); - } - private AlignedWritableMemChunk( List<IMeasurementSchema> schemaList, AlignedTVList list, boolean isTableModel, + IDeviceID physicalDeviceId, EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); this.measurementCategoryMap = new LinkedHashMap<>(); this.schemaList = new ArrayList<>(schemaList); this.useColumnSortedList = list instanceof ColumnSortedAlignedTVList; + this.physicalDeviceId = physicalDeviceId; this.sortColumnIndices = this.useColumnSortedList ? new ArrayList<>(((ColumnSortedAlignedTVList) list).getSortColumnIndices()) @@ -330,7 +361,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { !this.useColumnSortedList ? AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)) : AlignedTVListFactory.createFlatDeviceAlignedList( - new ArrayList<>(dataTypes), this.sortColumnIndices); + new ArrayList<>(dataTypes), this.sortColumnIndices, this.physicalDeviceId); this.dataTypes = this.list.getTsDataTypes(); } @@ -1407,8 +1438,8 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { list.serializeToWAL(buffer); } - public static AlignedWritableMemChunk deserialize(DataInputStream stream, boolean isTableModel) - throws IOException { + public static AlignedWritableMemChunk deserialize( + DataInputStream stream, boolean isTableModel, IDeviceID physicalDeviceId) throws IOException { int schemaListSize = stream.readInt(); List<IMeasurementSchema> schemaList = new ArrayList<>(schemaListSize); for (int i = 0; i < schemaListSize; i++) { @@ -1418,19 +1449,19 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { int sortedListSize = stream.readInt(); List<AlignedTVList> sortedList = new ArrayList<>(); for (int i = 0; i < sortedListSize; i++) { - AlignedTVList tvList = AlignedTVListFactory.deserialize(stream); + AlignedTVList tvList = AlignedTVListFactory.deserialize(stream, physicalDeviceId); sortedList.add(tvList); } - AlignedTVList list = AlignedTVListFactory.deserialize(stream); + AlignedTVList list = AlignedTVListFactory.deserialize(stream, physicalDeviceId); AlignedWritableMemChunk chunk = new AlignedWritableMemChunk( - schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); + schemaList, list, isTableModel, physicalDeviceId, EncryptUtils.getEncryptParameter()); chunk.sortedList = sortedList; return chunk; } public static AlignedWritableMemChunk deserializeSingleTVListMemChunks( - DataInputStream stream, boolean isTableModel) throws IOException { + DataInputStream stream, boolean isTableModel, IDeviceID physicalDeviceId) throws IOException { int schemaListSize = stream.readInt(); List<IMeasurementSchema> schemaList = new ArrayList<>(schemaListSize); for (int i = 0; i < schemaListSize; i++) { @@ -1438,9 +1469,9 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { schemaList.add(schema); } - AlignedTVList list = AlignedTVListFactory.deserialize(stream); + AlignedTVList list = AlignedTVListFactory.deserialize(stream, physicalDeviceId); return new AlignedWritableMemChunk( - schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); + schemaList, list, isTableModel, physicalDeviceId, EncryptUtils.getEncryptParameter()); } public List<IMeasurementSchema> getSchemaList() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java index 3bb49b9d499..05b87ed4e38 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java @@ -30,6 +30,8 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.EncryptUtils; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.schema.IMeasurementSchema; @@ -46,45 +48,20 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { private AlignedWritableMemChunk memChunk; - private EncryptParameter encryptParameter; - - public AlignedWritableMemChunkGroup(List<IMeasurementSchema> schemaList, boolean isTableModel) { - this(schemaList, isTableModel, 0); - } - - public AlignedWritableMemChunkGroup( - List<IMeasurementSchema> schemaList, boolean isTableModel, int sortColumnCount) { - this(schemaList, isTableModel, sortColumnCount, sortColumnCount > 0); - } - public AlignedWritableMemChunkGroup( List<IMeasurementSchema> schemaList, boolean isTableModel, List<Integer> sortColumnIndices, - boolean useColumnSortedList) { + boolean useColumnSortedList, + IDeviceID physicalDeviceId) { memChunk = new AlignedWritableMemChunk( - schemaList, isTableModel, sortColumnIndices, useColumnSortedList); - } - - public AlignedWritableMemChunkGroup( - List<IMeasurementSchema> schemaList, - boolean isTableModel, - int sortColumnCount, - boolean useColumnSortedList) { - this( - schemaList, - isTableModel, - createContiguousSortColumnIndices(sortColumnCount), - useColumnSortedList); - } - - private static List<Integer> createContiguousSortColumnIndices(int sortColumnCount) { - List<Integer> indices = new ArrayList<>(sortColumnCount); - for (int i = 0; i < sortColumnCount; i++) { - indices.add(i); - } - return indices; + schemaList, + isTableModel, + sortColumnIndices, + useColumnSortedList, + physicalDeviceId, + EncryptUtils.getEncryptParameter()); } @TestOnly @@ -234,14 +211,14 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { @Override public void setEncryptParameter(EncryptParameter encryptParameter) { - this.encryptParameter = encryptParameter; memChunk.setEncryptParameter(encryptParameter); } protected static AlignedWritableMemChunkGroup deserialize( - DataInputStream stream, boolean isTableModel) throws IOException { + DataInputStream stream, boolean isTableModel, IDeviceID physicalDeviceId) throws IOException { AlignedWritableMemChunkGroup memChunkGroup = new AlignedWritableMemChunkGroup(); - memChunkGroup.memChunk = AlignedWritableMemChunk.deserialize(stream, isTableModel); + memChunkGroup.memChunk = + AlignedWritableMemChunk.deserialize(stream, isTableModel, physicalDeviceId); return memChunkGroup; } @@ -251,10 +228,11 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { } protected static AlignedWritableMemChunkGroup deserializeSingleTVListMemChunks( - DataInputStream stream, boolean isTableModel) throws IOException { + DataInputStream stream, boolean isTableModel, IDeviceID physicalDeviceId) throws IOException { AlignedWritableMemChunkGroup memChunkGroup = new AlignedWritableMemChunkGroup(); memChunkGroup.memChunk = - AlignedWritableMemChunk.deserializeSingleTVListMemChunks(stream, isTableModel); + AlignedWritableMemChunk.deserializeSingleTVListMemChunks( + stream, isTableModel, physicalDeviceId); return memChunkGroup; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index fd52cc3b7e2..7541f2fa468 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -936,8 +936,11 @@ public class TsFileProcessor { && dataTypes[i].isBinary() && values != null && i < values.length - && values[i] != null) { - // Binary values keep their variable-length content outside the fixed Binary[] array. + && values[i] != null + && !(includeTagColumns + && columnCategories != null + && columnCategories[i] == TsTableColumnCategory.TAG)) { + // Flat TAG payload is counted once in the TVList dictionary memory correction. textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); } } @@ -1071,8 +1074,11 @@ public class TsFileProcessor { insertRowNode.isFlatDevice())) { continue; } - if (dataTypes[i].isBinary() && values[i] != null) { - // Binary values keep their variable-length content outside the fixed Binary[] array. + if (dataTypes[i].isBinary() + && values[i] != null + && !(insertRowNode.isFlatDevice() + && insertRowNode.getColumnCategories()[i] == TsTableColumnCategory.TAG)) { + // Flat TAG payload is counted once in the TVList dictionary memory correction. textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); } } @@ -1283,7 +1289,10 @@ public class TsFileProcessor { continue; } - if (dataType.isBinary()) { + if (dataType.isBinary() + && !(includeTagColumns + && columnCategories != null + && columnCategories[i] == TsTableColumnCategory.TAG)) { Binary[] binColumn = (Binary[]) columns[i]; // Binary values keep their variable-length content outside the fixed Binary[] arrays. // Failed rows are excluded because their values are not inserted into the memtable. 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 202e74ac833..2880e534865 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 @@ -63,11 +63,6 @@ public final class AlignedTVListFactory { } /** Creates flat-device storage with logical-device metadata shared by all TAG columns. */ - public static AlignedTVList createFlatDeviceAlignedList( - List<TSDataType> dataTypes, List<Integer> tagIndices) { - return new FlatDeviceAlignedTVList(dataTypes, tagIndices, true); - } - public static AlignedTVList createFlatDeviceAlignedList( List<TSDataType> dataTypes, List<Integer> tagIndices, IDeviceID physicalDeviceId) { Objects.requireNonNull(physicalDeviceId); @@ -85,18 +80,14 @@ public final class AlignedTVListFactory { byte encodedDataType = stream.readByte(); int dataTypeCount = stream.readInt(); if (encodedDataType == FlatDeviceAlignedTVList.WAL_DATA_TYPE_MARKER) { + Objects.requireNonNull(physicalDeviceId); return ColumnSortedAlignedTVList.deserialize( stream, dataTypeCount, true, - // null for test compatibility - physicalDeviceId == null ? "" : physicalDeviceId.getTableName()); - } - // 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) { - return ColumnSortedAlignedTVList.deserialize(stream, dataTypeCount); + physicalDeviceId.getTableName()); } + return AlignedTVList.deserialize( stream, TSDataType.deserialize(encodedDataType), dataTypeCount); } 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 50f762929a4..242fd968eb6 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 @@ -253,14 +253,9 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { serializeWALRows(buffer); } - @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 { + DataInputStream stream, int dataTypeCount, boolean flatDevice, String tableName) + throws IOException { List<TSDataType> dataTypes = new ArrayList<>(dataTypeCount); for (int columnIndex = 0; columnIndex < dataTypeCount; columnIndex++) { dataTypes.add(ReadWriteIOUtils.readDataType(stream)); @@ -272,7 +267,7 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { } ColumnSortedAlignedTVList tvList = flatDevice - ? new FlatDeviceAlignedTVList(dataTypes, sortColumnIndices, true) + ? new FlatDeviceAlignedTVList(dataTypes, sortColumnIndices, true, tableName) : new ColumnSortedAlignedTVList(dataTypes, sortColumnIndices, true); int rowCount = stream.readInt(); 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 49f13c06e67..202f9ac7287 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 @@ -485,7 +485,12 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { new MeasurementSchema("field", TSDataType.INT64), new MeasurementSchema("tag2", TSDataType.STRING)); AlignedWritableMemChunk mem = - new AlignedWritableMemChunk(schemas, true, List.of(0, 2), true); + new AlignedWritableMemChunk( + schemas, + true, + List.of(0, 2), + true, + IDeviceID.Factory.DEFAULT_FACTORY.create("flat_table")); for (String tag : List.of("b", "a")) { for (long time : new long[] {100, 1, 50}) { mem.writeAlignedPoints( 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 556558ee4a1..bd3e29674e9 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 @@ -35,6 +35,7 @@ import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.IDeviceID.Factory; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; @@ -182,7 +183,7 @@ public class FlatDeviceMemTableDeletionTest { assertFalse(buffer.hasRemaining()); AlignedWritableMemChunk recovered = AlignedWritableMemChunk.deserialize( - new DataInputStream(new ByteArrayInputStream(buffer.array())), true); + new DataInputStream(new ByteArrayInputStream(buffer.array())), true, Factory.DEFAULT_FACTORY.create(new String[] {"place_holder"})); try { recovered.getWorkingTVList().sort(); assertEquals(List.of(20L), remainingValues(recovered.getWorkingTVList(), 2)); @@ -234,7 +235,12 @@ public class FlatDeviceMemTableDeletionTest { new MeasurementSchema("tag1", TSDataType.STRING), new MeasurementSchema("field1", TSDataType.INT64))); AlignedWritableMemChunk chunk = - new AlignedWritableMemChunk(schemas, true, flat ? 1 : 0, flat); + new AlignedWritableMemChunk( + schemas, + true, + flat ? 1 : 0, + flat, + flat ? IDeviceID.Factory.DEFAULT_FACTORY.create(TABLE) : null); try { chunk.putAlignedRow(3, new Object[] {binary("b"), 30L}); chunk.putAlignedRow(1, new Object[] {binary("a"), 10L}); @@ -246,7 +252,9 @@ public class FlatDeviceMemTableDeletionTest { assertFalse(buffer.hasRemaining()); AlignedWritableMemChunk recovered = AlignedWritableMemChunk.deserialize( - new DataInputStream(new ByteArrayInputStream(buffer.array())), true); + new DataInputStream(new ByteArrayInputStream(buffer.array())), + true, + flat ? IDeviceID.Factory.DEFAULT_FACTORY.create(TABLE) : null); try { AlignedTVList list = recovered.getWorkingTVList(); assertTrue(list.isTimeDeleted(0)); 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 23a9c240b65..03d3188ffdb 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 @@ -69,6 +69,8 @@ public class FlatDeviceMemTableTest { @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); private static final String TABLE_NAME = "table1"; + private static final IDeviceID PHYSICAL_DEVICE = + IDeviceID.Factory.DEFAULT_FACTORY.create(TABLE_NAME); private static final String[] MEASUREMENTS = {"tag1", "tag2", "attr1", "field1"}; private static final TSDataType[] DATA_TYPES = { TSDataType.STRING, TSDataType.STRING, TSDataType.STRING, TSDataType.INT64 @@ -205,7 +207,8 @@ public class FlatDeviceMemTableTest { List<IMeasurementSchema> initialSchemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[3])); List<IMeasurementSchema> expandedSchemas = new ArrayList<>(List.of(SCHEMAS[1], tag3, SCHEMAS[3])); - AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(initialSchemas, true, 1, true); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(initialSchemas, true, 1, true, PHYSICAL_DEVICE); chunk.writeAlignedPoints(1, new Object[] {binary("a"), 10L}, new ArrayList<>(initialSchemas)); chunk.handoverAlignedTvList(); @@ -251,7 +254,8 @@ public class FlatDeviceMemTableTest { TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD }; - AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(initialSchemas, true, 1, true); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(initialSchemas, true, 1, true, PHYSICAL_DEVICE); chunk.writeAlignedPoints( 1, @@ -421,7 +425,8 @@ public class FlatDeviceMemTableTest { @Test public void testColumnSortedMemChunkWalRoundTrip() throws IOException { List<IMeasurementSchema> schemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[1], SCHEMAS[3])); - AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true, 2, true); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(schemas, true, 2, true, PHYSICAL_DEVICE); chunk.writeAlignedPoints( 1, new Object[] {binary("b"), binary("x"), 30L}, new ArrayList<>(schemas)); @@ -431,7 +436,7 @@ public class FlatDeviceMemTableTest { AlignedWritableMemChunk recovered = AlignedWritableMemChunk.deserialize( - new DataInputStream(new ByteArrayInputStream(buffer.array())), true); + new DataInputStream(new ByteArrayInputStream(buffer.array())), true, PHYSICAL_DEVICE); Assert.assertTrue(recovered.getWorkingTVList() instanceof ColumnSortedAlignedTVList); Assert.assertEquals( 2, ((ColumnSortedAlignedTVList) recovered.getWorkingTVList()).getSortColumnCount()); @@ -446,7 +451,8 @@ public class FlatDeviceMemTableTest { @Test public void testColumnSortedMemChunkFlushUsesCompoundRowKey() { List<IMeasurementSchema> schemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[3])); - AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true, 1, true); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(schemas, true, 1, true, PHYSICAL_DEVICE); chunk.putAlignedRow(1, new Object[] {binary("b"), 20L}); chunk.putAlignedRow(1, new Object[] {binary("a"), 10L}); chunk.putAlignedRow(1, new Object[] {binary("b"), null}); @@ -471,7 +477,8 @@ public class FlatDeviceMemTableTest { 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); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(schemas, true, 2, true, PHYSICAL_DEVICE); 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); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java index 5595390752a..320bb1503b8 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemChunkDeserializeTest.java @@ -33,6 +33,7 @@ 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.IDeviceID; +import org.apache.tsfile.file.metadata.IDeviceID.Factory; import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.read.TimeValuePair; @@ -261,7 +262,7 @@ public class MemChunkDeserializeTest { series.serializeToWAL(walBuffer); DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(walBuffer.getBuffer().array())); - AlignedWritableMemChunk memChunk = AlignedWritableMemChunk.deserialize(inputStream, false); + AlignedWritableMemChunk memChunk = AlignedWritableMemChunk.deserialize(inputStream, false, Factory.DEFAULT_FACTORY.create(new String[]{"place_holder"})); AlignedReadOnlyMemChunk readableChunk = (AlignedReadOnlyMemChunk) getAlignedReadOnlyChunk(memChunk, schemaList, measurementList); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java index 9d67c640acf..b12f8b1a70c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/TableDeletionEntryTest.java @@ -53,6 +53,54 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class TableDeletionEntryTest { + @Test + public void testFlatDeviceDeletionWithTableTimeIndex() throws IOException { + FlatDeviceDeletionEntry deletion = + new FlatDeviceDeletionEntry( + new DeletionPredicate("t", new SegmentExactMatch("a", 1)), new TimeRange(10, 20)); + deletion.setTagColumnNames(List.of("tag")); + ByteBuffer buffer = ByteBuffer.allocate(deletion.serializedSize()); + deletion.serialize(buffer); + assertEquals(deletion.serializedSize(), buffer.position()); + buffer.flip(); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + deletion.serialize(output); + // DataRegion checks the table-only resource key before allowing row-scoped TAG deletion. + // Both serialization paths and cloning must retain that distinction. + IDeviceID tableDevice = Factory.DEFAULT_FACTORY.create(new String[] {"t"}); + IDeviceID otherTable = Factory.DEFAULT_FACTORY.create(new String[] {"other"}); + for (ModEntry entry : + List.of( + deletion, + deletion.clone(), + ModEntry.createFrom(buffer), + ModEntry.createFrom(new ByteArrayInputStream(output.toByteArray())))) { + assertTrue(entry instanceof FlatDeviceDeletionEntry); + FlatDeviceDeletionEntry flatEntry = (FlatDeviceDeletionEntry) entry; + assertEquals(List.of("tag"), flatEntry.getTagColumnNames()); + assertTrue(entry.affects(tableDevice)); + assertTrue(entry.affects(tableDevice, 1, 10)); + assertTrue(entry.affects(tableDevice, 20, 30)); + assertTrue(entry.affects(tableDevice, 11, 19)); + assertFalse(entry.affects(tableDevice, 1, 9)); + assertFalse(entry.affects(tableDevice, 21, 30)); + assertFalse(entry.affects(otherTable, 10, 20)); + assertFalse(entry.affectsAll(tableDevice)); + assertTrue( + flatEntry.affects( + new Binary[] { + new Binary("t", TSFileConfig.STRING_CHARSET), + new Binary("a", TSFileConfig.STRING_CHARSET) + })); + assertFalse( + flatEntry.affects( + new Binary[] { + new Binary("t", TSFileConfig.STRING_CHARSET), + new Binary("b", TSFileConfig.STRING_CHARSET) + })); + } + } + @Test public void testChunkDeletionForNormalAndFlatDevices() { // Normal-table predicates are already matched against the chunk's logical device. Only diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/FlatDeviceAlignedChunkWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/FlatDeviceAlignedChunkWriterTest.java new file mode 100644 index 00000000000..ad7083843dd --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/FlatDeviceAlignedChunkWriterTest.java @@ -0,0 +1,118 @@ +/* + * 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.utils.EncryptDBUtils; + +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.statistics.Statistics; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.TsBlockBuilder; +import org.apache.tsfile.read.common.block.column.TimeColumn; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.write.chunk.ValueChunkWriter; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +public class FlatDeviceAlignedChunkWriterTest { + @Test + public void testTimeBatchOffsetAndRegressingStatistics() { + FlatDeviceAlignedChunkWriter writer = + new FlatDeviceAlignedChunkWriter(List.of(), EncryptDBUtils.getDefaultFirstEncryptParam()); + // Exclude the two sentinels; the batch endpoints are not its time bounds. + writer.write(new long[] {-999, 100, 1, 50, 999}, 3, 1); + assertStatistics(writer.getTimeChunkWriter().getPageWriter().getStatistics(), 3, 1, 100); + Assert.assertEquals(0, writer.getTimeChunkWriter().getNumOfPages()); + } + + @Test + public void testColumnRegionsAcrossPageBoundary() { + TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); + int oldMaxPoints = config.getMaxNumberOfPointsInPage(); + try { + config.setMaxNumberOfPointsInPage(3); + List<TSDataType> types = + List.of( + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.STRING); + List<IMeasurementSchema> schemas = new ArrayList<>(); + for (int i = 0; i < types.size(); i++) { + schemas.add(new MeasurementSchema("s" + i, types.get(i))); + } + FlatDeviceAlignedChunkWriter writer = + new FlatDeviceAlignedChunkWriter(schemas, EncryptDBUtils.getDefaultFirstEncryptParam()); + TsBlockBuilder builder = new TsBlockBuilder(types); + long[] times = {1000, 100, 1, 20, 30, 2}; + for (long time : times) { + builder.getTimeColumnBuilder().writeLong(time); + if (time <= 2) { + for (int column = 0; column < types.size(); column++) { + builder.getColumnBuilder(column).appendNull(); + } + } else { + builder.getColumnBuilder(0).writeBoolean(true); + builder.getColumnBuilder(1).writeInt((int) time); + builder.getColumnBuilder(2).writeLong(time); + builder.getColumnBuilder(3).writeFloat(time); + builder.getColumnBuilder(4).writeDouble(time); + builder + .getColumnBuilder(5) + .writeBinary(new Binary("v" + time, TSFileConfig.STRING_CHARSET)); + } + builder.declarePosition(); + } + TsBlock block = builder.build(); + // Both regions share backing arrays with an excluded sentinel. The second batch fills + // the first page, then continues at a nonzero offset into a page with regressing time. + writer.write((TimeColumn) block.getTimeColumn(), block.getValueColumns(), 2, 1); + Assert.assertEquals(0, writer.getTimeChunkWriter().getNumOfPages()); + writer.write((TimeColumn) block.getTimeColumn(), block.getValueColumns(), 3, 3); + Assert.assertEquals(1, writer.getTimeChunkWriter().getNumOfPages()); + assertStatistics(writer.getTimeChunkWriter().getStatistics(), 3, 1, 100); + assertStatistics(writer.getTimeChunkWriter().getPageWriter().getStatistics(), 2, 2, 30); + writer.sealCurrentPage(); + Assert.assertEquals(2, writer.getTimeChunkWriter().getNumOfPages()); + assertStatistics(writer.getTimeChunkWriter().getStatistics(), 5, 1, 100); + for (ValueChunkWriter value : writer.getValueChunkWriterList()) { + Assert.assertEquals(2, value.getNumOfPages()); + assertStatistics(value.getStatistics(), 3, 20, 100); + } + } finally { + config.setMaxNumberOfPointsInPage(oldMaxPoints); + } + } + + private void assertStatistics(Statistics<?> statistics, long count, long start, long end) { + Assert.assertEquals(count, statistics.getCount()); + Assert.assertEquals(start, statistics.getStartTime()); + Assert.assertEquals(end, statistics.getEndTime()); + } +}
