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 6aad990eb6228799f9ff8680be18bf3912cc03a3 Author: Tian Jiang <[email protected]> AuthorDate: Wed Oct 14 18:52:29 2026 +0800 tmp --- .../execution/operator/source/SeriesScanUtil.java | 8 +- .../utils/writer/AbstractCompactionWriter.java | 11 +- .../writer/ReadPointCrossCompactionWriter.java | 8 +- .../writer/RepairUnsortedFileCompactionWriter.java | 3 +- .../dataregion/memtable/TsFileProcessor.java | 9 +- .../wal/recover/file/TsFilePlanRedoer.java | 10 +- .../file/UnsealedTsFileRecoverPerformer.java | 16 + .../utils/writer/AbstractCompactionWriterTest.java | 43 ++- .../tablemodel/FlatDeviceCompactionTest.java | 72 ++++ .../dataregion/memtable/TsFileProcessorTest.java | 76 +++++ .../wal/recover/file/FlatDeviceRecoveryTest.java | 362 +++++++++++++++++++++ 11 files changed, 586 insertions(+), 32 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 566fc35f37a..e1a003e2f3a 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 @@ -34,7 +34,6 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; -import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.TagPredicateType; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskAlignedChunkLoader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskChunkLoader; @@ -880,12 +879,11 @@ public class SeriesScanUtil implements Accountable { private Filter getFlatDeviceDeletionFilter(TsFileResource resource) { List<FlatDeviceDeletionEntry> deletions; try { + // Flat timestamps restart between TAG groups. Even an unconditional deletion needs an + // independent row check because a page reader's monotonic deletion cursor can miss it. deletions = context.getPathModifications(resource, seriesPath.getDeviceId()).stream() - .filter( - mod -> - mod instanceof FlatDeviceDeletionEntry deletion - && deletion.getPredicate().getTagPredicateType() != TagPredicateType.NOP) + .filter(mod -> mod instanceof FlatDeviceDeletionEntry) .map(mod -> (FlatDeviceDeletionEntry) mod) .toList(); } catch (IllegalPathException e) { 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 15ecadd62b4..793c746e13f 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 @@ -124,26 +124,26 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { protected boolean isAlign; protected IDeviceID deviceId; - protected final Map<String, Set<String>> flatDeviceLogicalDevices = new HashMap<>(); + protected final Map<String, Set<IDeviceID>> flatDeviceLogicalDevices = new HashMap<>(); protected final Map<String, Long> flatDeviceRowCounts = new HashMap<>(); protected void recordFlatDeviceRows(TsBlock block, List<Integer> tagIndices) throws IOException { if (tagIndices.isEmpty()) { return; } - Set<String> devices = + Set<IDeviceID> devices = flatDeviceLogicalDevices.computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()); recordFlatDeviceLogicalDevices(block, tagIndices, devices); flatDeviceRowCounts.merge(deviceId.getTableName(), (long) block.getPositionCount(), Long::sum); } protected void recordFlatDeviceLogicalDevices( - TsBlock block, List<Integer> tagIndices, Set<String> devices) { + TsBlock block, List<Integer> tagIndices, Set<IDeviceID> devices) { recordFlatDeviceLogicalDevices(block, 0, block.getPositionCount(), tagIndices, devices); } protected void recordFlatDeviceLogicalDevices( - TsBlock block, int start, int count, List<Integer> tagIndices, Set<String> devices) { + TsBlock block, int start, int count, List<Integer> tagIndices, Set<IDeviceID> devices) { int end = start + count; for (int row = start; row < end; row++) { boolean newDevice = row == start; @@ -170,7 +170,8 @@ public abstract class AbstractCompactionWriter implements AutoCloseable { ? null : column.getBinary(row).getStringValue(TSFileConfig.STRING_CHARSET); } - devices.add(IDeviceID.Factory.DEFAULT_FACTORY.create(segments).toString()); + // Display strings can collide when TAG values contain dots; retain the segmented key. + devices.add(IDeviceID.Factory.DEFAULT_FACTORY.create(segments)); } } 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 8982c528395..f93fa004f21 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 @@ -35,6 +35,7 @@ import org.apache.tsfile.block.column.Column; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.file.header.PageHeader; import org.apache.tsfile.file.metadata.ChunkMetadata; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.block.TsBlock; @@ -51,7 +52,8 @@ 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, Set<IDeviceID>>> flatDevicesByFile = + new HashMap<>(); private final Map<TsFileIOWriter, Map<String, Long>> flatRowsByFile = new HashMap<>(); @TestOnly @@ -138,7 +140,7 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite estimateWrittenPointTotalSize(block, start, end - start); } TsFileIOWriter writer = targetFileWriters.get(targetIndex); - Set<String> devices = + Set<IDeviceID> devices = flatDevicesByFile .computeIfAbsent(writer, key -> new HashMap<>()) .computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()); @@ -157,7 +159,7 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite @Override protected void writeFlatDeviceStatistics(TsFileIOWriter writer) { - Map<String, Set<String>> tables = flatDevicesByFile.get(writer); + Map<String, Set<IDeviceID>> tables = flatDevicesByFile.get(writer); if (tables != null) { tables.forEach( (table, devices) -> 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 3b8e8c279e6..f17984910c5 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 @@ -113,8 +113,7 @@ public class RepairUnsortedFileCompactionWriter extends ReadPointInnerCompaction .computeIfAbsent(deviceId.getTableName(), key -> new HashSet<>()) .add( FlatDeviceCompactionUtils.getLogicalDevice( - deviceId, rowKeyColumnIndices, timeValuePair) - .toString()); + deviceId, rowKeyColumnIndices, timeValuePair)); flatDeviceRowCounts.merge(deviceId.getTableName(), 1L, Long::sum); } checkChunkSizeAndMayOpenANewChunk(fileWriter, chunkWriters[subTaskId], subTaskId); 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 8c01f490df1..d1c315b1576 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 @@ -217,7 +217,7 @@ public class TsFileProcessor { /** Flush file listener. */ private final List<FlushListener> flushListeners = new ArrayList<>(); - private final Map<String, Set<String>> flatDeviceLogicalDevices = new HashMap<>(); + private final Map<String, Set<IDeviceID>> flatDeviceLogicalDevices = new HashMap<>(); private final Map<String, Long> flatDeviceRowCounts = new HashMap<>(); private final QueryExecutionMetricSet QUERY_EXECUTION_METRICS = @@ -2217,9 +2217,10 @@ public class TsFileProcessor { alignedGroup.getAlignedMemChunk().getFlatDeviceEncodingStatistics(); if (statistics != null) { String table = device.getTableName(); - Set<String> devices = + Set<IDeviceID> devices = flatDeviceLogicalDevices.computeIfAbsent(table, key -> new HashSet<>()); - statistics.getDevices().forEach(id -> devices.add(id.toString())); + // DeviceID equality preserves segment boundaries; its display string does not. + devices.addAll(statistics.getDevices()); flatDeviceRowCounts.merge(table, statistics.getRowCount(), Long::sum); } } @@ -2231,7 +2232,7 @@ public class TsFileProcessor { if (logger.isDebugEnabled()) { logger.debug(StorageEngineMessages.START_TO_END_FILE, tsFileResource); } - for (Map.Entry<String, Set<String>> entry : flatDeviceLogicalDevices.entrySet()) { + for (Map.Entry<String, Set<IDeviceID>> entry : flatDeviceLogicalDevices.entrySet()) { TsFileTableSchemaUtil.addFlatDeviceStatistics( writer, entry.getKey(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java index 87f2ca54aef..c69cb940a86 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java @@ -91,7 +91,9 @@ public class TsFilePlanRedoer { return; } boolean isSingleDevice = !(node instanceof RelationalInsertTabletNode); - if (tsFileResource != null && isSingleDevice) { + // Flat resources index the whole table, not persisted logical rows. Replaying a row below + // that boundary is necessary; duplicate TAG/time keys are resolved by the aligned reader. + if (tsFileResource != null && isSingleDevice && !node.isFlatDevice()) { // orders of insert node is guaranteed by storage engine, just check time in the file // the last chunk group may contain the same data with the logs, ignore such logs in seq file Optional<Long> lastEndTime = @@ -123,7 +125,9 @@ public class TsFilePlanRedoer { for (Pair<IDeviceID, Integer> pair : deviceIndexPairs) { IDeviceID deviceID = pair.getLeft(); Optional<Long> endTimeInResource = - tsFileResource == null ? Optional.empty() : tsFileResource.getEndTime(deviceID); + tsFileResource == null || node.isFlatDevice() + ? Optional.empty() + : tsFileResource.getEndTime(deviceID); long minTimeOfDevice = relationalInsertTabletNode.getTimes()[start]; if (endTimeInResource.isPresent() && endTimeInResource.get() >= minTimeOfDevice) { start = pair.getRight(); @@ -154,7 +158,7 @@ public class TsFilePlanRedoer { if (!node.hasValidMeasurements()) { continue; } - if (tsFileResource != null) { + if (tsFileResource != null && !node.isFlatDevice()) { // orders of insert node is guaranteed by storage engine, just check time in the file // the last chunk group may contain the same data with the logs, ignore such logs in seq // file diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java index 0ae6e38e5f6..834da656189 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java @@ -32,13 +32,16 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDe import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.storageengine.dataregion.flush.CompressionRatio; import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask; +import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunkGroup; import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunkGroup; 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.ModEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.FullExactMatch; +import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.TagPredicateType; import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndexCacheRecorder; @@ -137,6 +140,12 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform long endTime = chunkMetaData.getEndTime(); // exist deletion for current measurement for (ModEntry modification : modificationsForResource) { + // A TAG predicate can delete only some logical devices sharing this physical chunk. + // Without row keys, keep conservative bounds and defer the predicate to row filtering. + if (modification instanceof FlatDeviceDeletionEntry deletion + && deletion.getPredicate().getTagPredicateType() != TagPredicateType.NOP) { + continue; + } if (!(modification.affects(deviceId, startTime, endTime) && modification.affects(chunkMetaData.getMeasurementUid()))) { continue; @@ -182,6 +191,13 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform if (tsFileResource != null) { // delete data already flushed in the MemTable to avoid duplicates for (IDeviceID device : tsFileResource.getDevices()) { + IWritableMemChunkGroup group = memTable.getMemTableMap().get(device); + if (group instanceof AlignedWritableMemChunkGroup alignedGroup + && alignedGroup.getAlignedMemChunk().isFlatDevice()) { + // Table bounds cannot prove that any particular snapshot row was flushed. + // Keep flat rows for replay, including devices whose time restarts earlier. + continue; + } // iterating the index, must present if (device.isTableModel()) { memTable.delete( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java index 11035a2a651..b544c2a14e5 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriterTest.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.exe import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.enums.TSDataType; @@ -31,8 +32,10 @@ import org.apache.tsfile.exception.write.PageException; import org.apache.tsfile.file.header.PageHeader; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.file.metadata.statistics.Statistics; 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.write.chunk.ChunkWriterImpl; @@ -45,6 +48,14 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collections; import java.util.List; +import java.util.Map; + +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class AbstractCompactionWriterTest { @@ -54,19 +65,25 @@ public class AbstractCompactionWriterTest { public void testCrossCompactionUsesExplicitFlatDeviceProperty() throws IOException { IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1"}); TsFileResource source = new TsFileResource(); - source.setTimeIndex(new ArrayDeviceTimeIndex()); + source.setTimeIndex(new FileTimeIndex(1, 100)); // A normal table with all-null TAGs also has a single-segment device ID. Its device - // boundary must not be replaced by the aggregate boundary of the entire table. - source.updateStartTime(device, 1); - source.updateEndTime(device, 20); - source.updateTableStartTime("table1", 1); - source.updateTableEndTime("table1", 100); + // boundary must be read from metadata when the resource has only file-level bounds. + TsFileSequenceReader reader = mock(TsFileSequenceReader.class); + TimeseriesMetadata metadata = new TimeseriesMetadata(); + Statistics<?> statistics = Statistics.getStatsByType(TSDataType.INT64); + statistics.update(1, 1L); + statistics.update(20, 2L); + metadata.setStatistics(statistics); + when(reader.readDeviceMetadata(device)).thenReturn(Map.of("s1", metadata)); try (ReadPointCrossCompactionWriter writer = - new ReadPointCrossCompactionWriter( - Collections.emptyList(), Collections.singletonList(source))) { + spy( + new ReadPointCrossCompactionWriter( + Collections.emptyList(), Collections.singletonList(source)))) { + doReturn(reader).when(writer).getFileReader(source); writer.startChunkGroup(device, true, true); Assert.assertEquals(100, writer.currentDeviceEndTime[0]); Assert.assertTrue(writer.isCurrentDeviceExistedInSourceSeqFiles[0]); + verify(reader, never()).readDeviceMetadata(device); writer.startChunkGroup(device, true, false); Assert.assertEquals(20, writer.currentDeviceEndTime[0]); @@ -78,12 +95,18 @@ public class AbstractCompactionWriterTest { } @Test - public void testCrossCompactionFlatDeviceDoesNotRequireDeviceTimeIndexEntry() throws IOException { + public void testCrossCompactionFlatDeviceUsesMergedTableTimeIndexEntry() throws IOException { IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1"}); + IDeviceID logicalDevice = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1", "tag1"}); TsFileResource source = new TsFileResource(); source.setTimeIndex(new ArrayDeviceTimeIndex()); source.updateTableStartTime("table1", 1); source.updateTableEndTime("table1", 100); + // Flat table bounds share the ordinary time index under the table-only physical ID. + // No logical TAG device entry is required for the flat branch. + Assert.assertTrue(source.getTimeIndex().checkDeviceIdExist(device)); + Assert.assertFalse(source.getTimeIndex().checkDeviceIdExist(logicalDevice)); try (ReadPointCrossCompactionWriter writer = new ReadPointCrossCompactionWriter( Collections.emptyList(), Collections.singletonList(source))) { @@ -91,7 +114,7 @@ public class AbstractCompactionWriterTest { Assert.assertEquals(100, writer.currentDeviceEndTime[0]); Assert.assertTrue(writer.isCurrentDeviceExistedInSourceSeqFiles[0]); - writer.startChunkGroup(device, true, false); + writer.startChunkGroup(logicalDevice, true, false); Assert.assertEquals(Long.MIN_VALUE, writer.currentDeviceEndTime[0]); Assert.assertFalse(writer.isCurrentDeviceExistedInSourceSeqFiles[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 6f4da013a8e..25befbc94e2 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 @@ -100,6 +100,78 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { assertFlatDeviceRowsSurviveCompaction(new ReadPointCompactionPerformer()); } + @Test + public void testInnerCompactionStatisticsKeepTagSegmentBoundaries() throws Exception { + seqResources.add( + createFlatDeviceFile(true, 10, new String[] {"a", "a.b"}, new String[] {"b.c", "c"}, 1)); + seqResources.add( + createFlatDeviceFile(true, 20, new String[] {"a", "a.b"}, new String[] {"b.c", "c"}, 3)); + tsFileManager.addAll(seqResources, true); + Assert.assertTrue( + new InnerSpaceCompactionTask( + 0, tsFileManager, seqResources, true, new ReadPointCompactionPerformer(), 0) + .start()); + TsFileResource target = tsFileManager.getTsFileList(true).get(0); + assertFlatDeviceStatistics(target, 2, 4); + Assert.assertEquals(4, readFlatDeviceRows(target).size()); + } + + @Test + public void testCrossCompactionStatisticsKeepTagSegmentBoundaries() throws Exception { + // Every target contains two distinct TAG tuples that both display as flat_table.a.b.c. + // Repeated source keys are overwritten rather than counted as additional devices or rows. + for (long time : new long[] {1, 3}) { + seqResources.add( + createFlatDeviceFile( + true, 10, new String[] {"a", "a.b"}, new String[] {"b.c", "c"}, time)); + unseqResources.add( + createFlatDeviceFile( + false, 20, new String[] {"a", "a.b"}, new String[] {"b.c", "c"}, time)); + } + tsFileManager.addAll(seqResources, true); + tsFileManager.addAll(unseqResources, false); + Assert.assertTrue( + new CrossSpaceCompactionTask( + 0, + tsFileManager, + seqResources, + unseqResources, + new ReadPointCompactionPerformer(), + 0, + 0) + .start()); + Assert.assertEquals(2, tsFileManager.getTsFileList(true).size()); + for (TsFileResource target : tsFileManager.getTsFileList(true)) { + assertFlatDeviceStatistics(target, 2, 2); + Assert.assertEquals(2, readFlatDeviceRows(target).size()); + } + } + + @Test + public void testRepairStatisticsKeepTagSegmentBoundaries() throws Exception { + TsFileResource source = createEmptyFileAndResource(true); + seqResources.add(source); + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(source)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", Collections.emptyList()); + writer.generateFlatDeviceAlignedRows( + new long[] {1, 1, 1}, + new String[] {"a.b", "a", "a.b"}, + new long[] {10, 20, 10}, + new String[] {"c", "b.c", "c"}); + writer.endChunkGroup(); + writer.endFile(); + } + tsFileManager.addAll(seqResources, true); + source.setTsFileRepairStatus(TsFileRepairStatus.NEED_TO_REPAIR_BY_REWRITE); + Assert.assertTrue( + new RepairUnsortedFileCompactionTask(0, tsFileManager, source, true, 0).start()); + TsFileResource target = tsFileManager.getTsFileList(false).get(0); + assertFlatDeviceStatistics(target, 2, 2); + Assert.assertEquals(2, readFlatDeviceRows(target).size()); + } + @Test public void testInnerUnseqCompactionWithOverlappingFlatDeviceFiles() throws Exception { assertInnerUnseqFlatDeviceCompaction(3, 20, 8, 37, 11, false); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 16ffa8761ad..e8012404964 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -1502,6 +1502,82 @@ public class TsFileProcessorTest { assertFlatDeviceResourceTimes(false); } + @Test + public void testFlatDeviceStatisticsKeepTagSegmentBoundaries() throws Exception { + processor = + new TsFileProcessor( + storageGroup, + new File(filePath), + sgInfo, + this::closeTsFileProcessor, + (tsFileProcessor, updateMap, systemFlushTime) -> {}, + false); + processor.setTsFileProcessorInfo(new TsFileProcessorInfo(sgInfo)); + sgInfo.initTsFileProcessorInfo(processor); + SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor); + MeasurementSchema[] schemas = { + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("tag2", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT64) + }; + processor.registerToTsFile( + "flat_table", + table -> + new TableSchema( + table, + Arrays.asList(schemas), + List.of(ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD)), + true); + // These devices have the same display string but distinct TAG segments. Exercise both + // deduplication inside one flush and the device union across multiple flushes. + for (long time : new long[] {10, 20}) { + for (String[] tags : new String[][] {{"a.b", "c"}, {"a", "b.c"}}) { + RelationalInsertRowNode row = + new RelationalInsertRowNode( + new PlanNodeId("row"), + new PartialPath("flat_table", false), + true, + new String[] {"tag1", "tag2", "s1"}, + new TSDataType[] {TSDataType.STRING, TSDataType.STRING, TSDataType.INT64}, + schemas, + time, + new Object[] { + new Binary(tags[0], StandardCharsets.UTF_8), + new Binary(tags[1], StandardCharsets.UTF_8), + time + }, + false, + new TsTableColumnCategory[] { + TsTableColumnCategory.TAG, TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD + }); + row.setFlatDevice(true); + processor.insert(row, new long[5]); + } + processor.asyncFlush(); + processor.getCloseFuture().get(30, TimeUnit.SECONDS); + processor.setCloseFuture(null); + } + processor.asyncClose().get(30, TimeUnit.SECONDS); + try (TsFileSequenceReader reader = new TsFileSequenceReader(filePath)) { + assertEquals( + 2, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get( + TsFileTableSchemaUtil.getFlatDeviceLogicalDeviceCountPropertyKey( + "flat_table"))) + .getLong()); + assertEquals( + 4, + ByteBuffer.wrap( + reader + .getTsFileProperties() + .get(TsFileTableSchemaUtil.getFlatDevicePointCountPropertyKey("flat_table"))) + .getLong()); + } + } + private void assertFlatDeviceResourceTimes(boolean sequence) throws Exception { Map<IDeviceID, Long> flushedTimes = new HashMap<>(); processor = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java new file mode 100644 index 00000000000..f272be0df21 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java @@ -0,0 +1,362 @@ +/* + * 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.wal.recover.file; + +import org.apache.iotdb.commons.path.AlignedFullPath; +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.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; +import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; +import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; +import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable; +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.TagPredicate; +import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.TsFileUtilsForRecoverTest; +import org.apache.iotdb.db.utils.EncryptDBUtils; +import org.apache.iotdb.db.utils.EnvironmentUtils; + +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.enums.CompressionType; +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.read.reader.IPointReader; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.io.File; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Set; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class FlatDeviceRecoveryTest { + private static final String DATABASE = "root.flat_recovery"; + private static final String TABLE = "flat_table"; + private static final IDeviceID PHYSICAL_DEVICE = + IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {TABLE}); + private static final MeasurementSchema[] SCHEMAS = { + new MeasurementSchema("tag", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("s", TSDataType.INT64) + }; + private File file; + private TsFileResource resource; + + @Before + public void setUp() throws Exception { + EnvironmentUtils.envSetUp(); + file = new File(TsFileUtilsForRecoverTest.getTestTsFilePath(DATABASE, 0, 0, 1)); + Files.createDirectories(file.getParentFile().toPath()); + resource = new TsFileResource(file); + resource.setSeq(false); + TsTable table = new TsTable(TABLE); + table.addColumnSchema(new TagColumnSchema("tag", TSDataType.STRING)); + table.addColumnSchema( + new FieldColumnSchema( + "s", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.UNCOMPRESSED)); + table.addProp(TsTable.FLAT_DEVICE_PROPERTY, "true"); + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, TABLE, null); + } + + @After + public void tearDown() throws Exception { + if (resource != null) { + resource.close(); + } + EnvironmentUtils.cleanDir(file.getParent()); + EnvironmentUtils.cleanEnv(); + } + + @Test + public void testRowReplayBelowTableEndTime() throws Exception { + setTableBounds(); + TsFilePlanRedoer redoer = new TsFilePlanRedoer(resource); + redoer.redoInsert(row("B", 50, 2)); + assertEquals(List.of("50:B:2"), readMemTable(redoer.getRecoveryMemTable())); + } + + @Test + public void testBatchRowReplayBelowTableEndTime() throws Exception { + setTableBounds(); + TsFilePlanRedoer redoer = new TsFilePlanRedoer(resource); + redoer.redoInsertRows( + new RelationalInsertRowsNode( + new PlanNodeId("rows"), List.of(0, 1), List.of(row("B", 50, 2), row(null, 60, 3)))); + assertEquals(List.of("60:null:3", "50:B:2"), readMemTable(redoer.getRecoveryMemTable())); + } + + @Test + public void testTabletReplayIncludesNullTagBelowTableEndTime() throws Exception { + setTableBounds(); + // The null-TAG logical ID equals the physical table ID, so even a logical-ID lookup + // must not use the table's end time to skip this device's unflushed tablet rows. + BitMap tagNulls = new BitMap(2); + tagNulls.mark(0); + RelationalInsertTabletNode tablet = + new RelationalInsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath(TABLE, false), + true, + new String[] {"tag", "s"}, + new TSDataType[] {TSDataType.STRING, TSDataType.INT64}, + SCHEMAS, + new long[] {50, 60}, + new BitMap[] {tagNulls, null}, + new Object[] {new Binary[] {null, binary("B")}, new long[] {2, 3}}, + 2, + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + tablet.setFlatDevice(true); + TsFilePlanRedoer redoer = new TsFilePlanRedoer(resource); + redoer.redoInsert(tablet); + assertEquals(List.of("50:null:2", "60:B:3"), readMemTable(redoer.getRecoveryMemTable())); + } + + @Test + public void testSnapshotRetainsRowsInsideFlushedTableRange() throws Exception { + writeUnsealedFile(); + IMemTable snapshot = new PrimitiveMemTable(DATABASE, "0"); + snapshot.insertAlignedRow(row("B", 50, 2)); + snapshot.insertAlignedRow(row("A", 100, 3)); + try (UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer(resource, false, ignored -> {})) { + performer.startRecovery(); + performer.redoLog(new WALInfoEntry(1, snapshot)); + assertEquals(List.of("100:A:3", "50:B:2"), readMemTable(snapshot)); + // Verify that the snapshot is actually installed in the redoer, not just left untouched. + performer.redoLog(new WALInfoEntry(1, row("C", 60, 4))); + assertEquals(List.of("100:A:3", "50:B:2", "60:C:4"), readMemTable(snapshot)); + performer.endRecovery(); + } + assertEquals(List.of("10:A:1", "100:A:3", "20:B:2", "50:B:2", "60:C:4"), readRecoveredFile()); + } + + @Test + public void testCrashRecoveryReplaysAndMergesDuplicateFlatKeys() throws Exception { + writeUnsealedFile(); + try (UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer(resource, false, ignored -> {})) { + performer.startRecovery(); + // A duplicate already exists on disk, while new logical rows restart below table time 100. + // The later WAL value must win and each full TAG/time key must appear only once. + performer.redoLog(new WALInfoEntry(1, row("A", 100, 3))); + performer.redoLog(new WALInfoEntry(1, row("B", 50, 4))); + performer.redoLog(new WALInfoEntry(1, row(null, 60, 5))); + performer.endRecovery(); + } + assertEquals( + List.of("60:null:5", "10:A:1", "100:A:3", "20:B:2", "50:B:4"), readRecoveredFile()); + } + + @Test + public void testTagDeletionCoveringChunkKeepsResourceBounds() throws Exception { + assertRecoveredBounds( + new TagPredicate.SegmentExactMatch("A", 1), + new TimeRange(0, 200), + 10, + 100, + List.of("20:B:2")); + } + + @Test + public void testTagDeletionTouchingChunkBoundaryKeepsResourceBounds() throws Exception { + assertRecoveredBounds( + new TagPredicate.SegmentExactMatch("A", 1), + new TimeRange(0, 30), + 10, + 100, + List.of("100:A:1", "20:B:2")); + } + + @Test + public void testUnconditionalDeletionStillTrimsResourceBounds() throws Exception { + assertRecoveredBounds( + new TagPredicate.NOP(), new TimeRange(0, 30), 31, 100, List.of("100:A:1")); + } + + private void assertRecoveredBounds( + TagPredicate predicate, TimeRange range, long start, long end, List<String> expectedRows) + throws Exception { + writeUnsealedFile(); + FlatDeviceDeletionEntry deletion = + new FlatDeviceDeletionEntry(new DeletionPredicate(TABLE, predicate), range); + deletion.setTagColumnNames(List.of("tag")); + resource.getModFileForWrite().write(List.of(deletion)); + resource.getModFileForWrite().close(); + try (UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer(resource, false, ignored -> {})) { + performer.startRecovery(); + assertTrue(performer.hasCrashed()); + assertEquals(start, resource.getStartTime(PHYSICAL_DEVICE).get().longValue()); + assertEquals(end, resource.getEndTime(PHYSICAL_DEVICE).get().longValue()); + if (!(predicate instanceof TagPredicate.NOP)) { + assertTrue(resource.isSatisfied(PHYSICAL_DEVICE, TimeFilterApi.eq(20), false, false)); + } + performer.endRecovery(); + } + assertEquals(expectedRows, readRecoveredFile()); + } + + private void setTableBounds() { + resource.updateTableStartTime(TABLE, 10); + resource.updateTableEndTime(TABLE, 100); + } + + private void writeUnsealedFile() throws Exception { + try (TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer + .getSchema() + .registerTableSchema( + TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute( + DataNodeTableCache.getInstance().getTable(DATABASE, TABLE))); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, TABLE, true); + writer.startChunkGroup(PHYSICAL_DEVICE); + FlatDeviceAlignedChunkWriter chunk = + new FlatDeviceAlignedChunkWriter( + Arrays.asList(SCHEMAS), EncryptDBUtils.getDefaultFirstEncryptParam()); + for (RelationalInsertRowNode row : + List.of(row("A", 10, 1), row("A", 100, 1), row("B", 20, 2))) { + chunk.write(row.getTime(), (Binary) row.getValues()[0], false); + chunk.write(row.getTime(), (long) row.getValues()[1], false); + chunk.write(row.getTime()); + } + chunk.writeToFileWriter(writer); + writer.endChunkGroup(); + // A completed flush writes this marker; leave the footer unwritten to simulate a crash. + writer.writePlanIndices(); + } + } + + private RelationalInsertRowNode row(String tag, long time, long value) throws Exception { + RelationalInsertRowNode row = + new RelationalInsertRowNode( + new PlanNodeId("row"), + new PartialPath(TABLE, false), + true, + new String[] {"tag", "s"}, + new TSDataType[] {TSDataType.STRING, TSDataType.INT64}, + SCHEMAS, + time, + new Object[] {tag == null ? null : binary(tag), value}, + false, + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + row.setFlatDevice(true); + return row; + } + + private List<String> readMemTable(IMemTable memTable) throws Exception { + AlignedFullPath path = + new AlignedFullPath(PHYSICAL_DEVICE, List.of("tag", "s"), Arrays.asList(SCHEMAS)); + IPointReader reader = + memTable + .query(new QueryContext(false, false), path, Long.MIN_VALUE, null, null) + .getPointReader(); + List<String> rows = new ArrayList<>(); + while (reader.hasNextTimeValuePair()) { + TimeValuePair row = reader.nextTimeValuePair(); + var values = row.getValue().getVector(); + rows.add( + row.getTimestamp() + + ":" + + (values[0] == null ? "null" : values[0].getBinary()) + + ":" + + values[1].getLong()); + } + return rows; + } + + private static Binary binary(String value) { + return new Binary(value, TSFileConfig.STRING_CHARSET); + } + + private List<String> readRecoveredFile() throws Exception { + // DataRegion's recovery callback closes a sealed resource before exposing it to queries. + // Mirror that lifecycle so the scan reads the recovered footer instead of memory metadata. + resource.close(); + SeriesScanOptions.Builder options = new SeriesScanOptions.Builder(); + options.withAllSensors(Set.of("tag", "s")); + options.withGlobalTimeFilter(TimeFilterApi.gt(0)); + FragmentInstanceContext context = + FragmentInstanceContext.createFragmentInstanceContextForCompaction(0L); + // Table scans preserve rows with all-null TAGs, including the key-only page projection. + context.setIgnoreAllNullRows(false); + AlignedSeriesScanUtil scan = + new AlignedSeriesScanUtil( + new AlignedFullPath(PHYSICAL_DEVICE, List.of("tag", "s"), Arrays.asList(SCHEMAS)), + Ordering.ASC, + options.build(), + context, + true, + List.of(TSDataType.STRING, TSDataType.INT64), + List.of(0)); + scan.initQueryDataSource(new QueryDataSource(List.of(), List.of(resource))); + List<String> rows = new ArrayList<>(); + while (scan.hasNextFile().orElse(true)) { + while (scan.hasNextChunk().orElse(true)) { + while (scan.hasNextPage()) { + TsBlock block = scan.nextPage(); + if (block == null) { + continue; + } + for (int rowIndex = 0; rowIndex < block.getPositionCount(); rowIndex++) { + rows.add( + block.getTimeByIndex(rowIndex) + + ":" + + (block.getColumn(0).isNull(rowIndex) + ? "null" + : block.getColumn(0).getBinary(rowIndex)) + + ":" + + block.getColumn(1).getLong(rowIndex)); + } + } + } + } + return rows; + } +}
