From 0b7733ded585a39441988358d8ac29e40b9a7946 Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Wed, 14 Oct 2026 17:27:44 +0800 Subject: [PATCH 1/2] Fix deletion WAL logging and recovery during memtable flush --- .../dataregion/memtable/TsFileProcessor.java | 94 ++++--- .../wal/recover/WALNodeRecoverTask.java | 16 +- .../memtable/TsFileProcessorTest.java | 242 ++++++++++++++++++ .../wal/recover/WALRecoverManagerTest.java | 72 ++++++ 4 files changed, 386 insertions(+), 38 deletions(-) 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 cc931e75fac10..fda5bad3bb94d 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 @@ -1626,13 +1626,35 @@ public boolean deleteDataInMemory(ModEntry deletion) { } public WALFlushListener logDeleteDataNodeInWAL(DeleteDataNode deleteDataNode) { - walEntryNum++; - return walNode.log(workMemTable.getMemTableId(), deleteDataNode); + flushQueryLock.readLock().lock(); + try { + walEntryNum++; + return walNode.log(getMemTableIdForDeletion(), deleteDataNode); + } finally { + flushQueryLock.readLock().unlock(); + } } public WALFlushListener logDeleteDataNodeInWAL(RelationalDeleteDataNode deleteDataNode) { - walEntryNum++; - return walNode.log(workMemTable.getMemTableId(), deleteDataNode); + flushQueryLock.readLock().lock(); + try { + walEntryNum++; + return walNode.log(getMemTableIdForDeletion(), deleteDataNode); + } finally { + flushQueryLock.readLock().unlock(); + } + } + + // Called with flushQueryLock held until the deletion is queued in WAL, before the selected + // memtable can be released and its flush checkpoint recorded. + private long getMemTableIdForDeletion() { + if (workMemTable != null) { + return workMemTable.getMemTableId(); + } + IMemTable flushingMemTable = flushingMemTables.peekLast(); + // A pending flush still needs this deletion during recovery. If no memtable remains, keep + // logging with the existing sentinel so that IoTConsensus can replicate the deletion. + return flushingMemTable == null ? MEMTABLE_NOT_EXIST : flushingMemTable.getMemTableId(); } public TsFileResource getTsFileResource() { @@ -1983,39 +2005,45 @@ public void flushOneMemTable() { } } - try { + // Keep deletion registration atomic with writing modifications and releasing the memtable. + // Acquire the deque monitor first, matching syncReleaseFlushedMemTable's lock order. + synchronized (flushingMemTables) { flushQueryLock.writeLock().lock(); - Iterator> iterator = modsToMemtable.iterator(); - while (iterator.hasNext()) { - Pair entry = iterator.next(); - if (entry.right.equals(memTableToFlush)) { - this.tsFileResource.getModFileForWrite().write(entry.left); - tsFileResource.getModFileForWrite().close(); - iterator.remove(); - logger.info(StorageEngineMessages.DELETION_WRITTEN_WHEN_FLUSH, entry.left); + try { + try { + Iterator> iterator = modsToMemtable.iterator(); + while (iterator.hasNext()) { + Pair entry = iterator.next(); + if (entry.right.equals(memTableToFlush)) { + this.tsFileResource.getModFileForWrite().write(entry.left); + tsFileResource.getModFileForWrite().close(); + iterator.remove(); + logger.info(StorageEngineMessages.DELETION_WRITTEN_WHEN_FLUSH, entry.left); + } + } + } catch (IOException e) { + logger.error( + StorageEngineMessages + .STORAGE_LOG_MEET_ERROR_WHEN_WRITING_INTO_MODIFICATIONFILE_FILE_OF_63B5E24A, + tsFileResource.getTsFile().getAbsolutePath(), + e); + DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e); } - } - } catch (IOException e) { - logger.error( - StorageEngineMessages - .STORAGE_LOG_MEET_ERROR_WHEN_WRITING_INTO_MODIFICATIONFILE_FILE_OF_63B5E24A, - tsFileResource.getTsFile().getAbsolutePath(), - e); - DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e); - } finally { - flushQueryLock.writeLock().unlock(); - } - if (logger.isDebugEnabled()) { - logger.debug( - StorageEngineMessages.STORAGE_LOG_TRY_GET_LOCK_TO_RELEASE_A_MEMTABLE_SIGNAL_B9098E21, - dataRegionName, - tsFileResource.getTsFile().getAbsolutePath(), - memTableToFlush.isSignalMemTable()); - } + if (logger.isDebugEnabled()) { + logger.debug( + StorageEngineMessages.STORAGE_LOG_TRY_GET_LOCK_TO_RELEASE_A_MEMTABLE_SIGNAL_B9098E21, + dataRegionName, + tsFileResource.getTsFile().getAbsolutePath(), + memTableToFlush.isSignalMemTable()); + } - // For sync flush - syncReleaseFlushedMemTable(memTableToFlush); + // For sync flush + syncReleaseFlushedMemTable(memTableToFlush); + } finally { + flushQueryLock.writeLock().unlock(); + } + } try { writer.getTsFileOutput().force(); } catch (IOException e) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java index 7efce7036563b..30fdf97c414bc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java @@ -54,6 +54,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicLong; @@ -73,6 +74,8 @@ public class WALNodeRecoverTask implements Runnable { private long firstValidVersionId = Long.MAX_VALUE; private Map memTableId2Info; private Map memTableId2RecoverPerformer; + // Multiple unflushed memtables in one TsFile share a single recovery lifecycle. + private final Set recoverPerformers = new HashSet<>(); public WALNodeRecoverTask(File logDirectory, CountDownLatch allNodesRecoveredLatch) { this.logDirectory = logDirectory; @@ -92,11 +95,11 @@ public void run() { recoverInfoFromCheckpoints(); recoverTsFiles(); } catch (Exception e) { - for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { recoverPerformer.getRecoverListener().fail(e); } } finally { - for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { try { if (!recoverPerformer.canWrite()) { recoverPerformer.close(); @@ -263,13 +266,16 @@ private void recoverInfoFromCheckpoints() { } // update firstValidVersionId and get recover performer from WALRecoverManager memTableId2RecoverPerformer = new HashMap<>(); + Map file2RecoverPerformer = new HashMap<>(); for (MemTableInfo memTableInfo : memTableId2Info.values()) { firstValidVersionId = Math.min(firstValidVersionId, memTableInfo.getFirstFileVersionId()); + File tsFile = new File(memTableInfo.getTsFilePath()); UnsealedTsFileRecoverPerformer recoverPerformer = - walRecoverManger.removeRecoverPerformer(new File(memTableInfo.getTsFilePath())); + file2RecoverPerformer.computeIfAbsent(tsFile, walRecoverManger::removeRecoverPerformer); if (recoverPerformer != null) { memTableId2RecoverPerformer.put(memTableInfo.getMemTableId(), recoverPerformer); + recoverPerformers.add(recoverPerformer); } } } @@ -279,7 +285,7 @@ private void recoverTsFiles() { return; } // make preparation for recovery - for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { try { recoverPerformer.startRecovery(); } catch (Exception e) { @@ -348,7 +354,7 @@ private void recoverTsFiles() { private void endRecovery() { // end recovering all recover performers - for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { try { recoverPerformer.endRecovery(); recoverPerformer.getRecoverListener().succeed(); 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 b4117ec121a94..29c8bed8b0efd 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 @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.file.SystemFileFactory; import org.apache.iotdb.commons.path.AlignedFullPath; import org.apache.iotdb.commons.path.AlignedPath; +import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; @@ -35,14 +36,27 @@ import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo; import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest; +import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; +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.wal.WALManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader; +import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener; import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; @@ -62,6 +76,7 @@ import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Path; import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.read.expression.QueryExpression; import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.read.reader.IPointReader; @@ -80,6 +95,8 @@ import java.io.File; import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; @@ -87,10 +104,13 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static junit.framework.TestCase.assertTrue; import static org.apache.iotdb.db.storageengine.dataregion.DataRegionTest.buildInsertRowNodeByTSRecord; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -198,6 +218,228 @@ public void testRegisterToTsFileCachesTableSchemaByVersion() throws IOException updatedSchema, processor.getWriter().getSchema().getTableSchemaMap().get("table1")); } + @Test + public void testDeletionDuringFlushCompletionIsPersisted() throws Exception { + WALMode originalWalMode = config.getWalMode(); + config.setWalMode(WALMode.SYNC); + try { + processor = + new TsFileProcessor( + storageGroup, + SystemFileFactory.INSTANCE.getFile(filePath), + sgInfo, + this::closeTsFileProcessor, + (tsFileProcessor, updateMap, systemFlushTime) -> {}, + true); + processor.setTsFileProcessorInfo(new TsFileProcessorInfo(sgInfo)); + sgInfo.initTsFileProcessorInfo(processor); + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(DataPoint.getDataPoint(dataType, measurementId, "1")); + processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); + processor.setManagedByFlushManager(true); + processor.asyncFlush(); + Assert.assertNull(processor.getWorkMemTable()); + + MeasurementPath path = new MeasurementPath(deviceId + "." + measurementId); + TreeDeletionEntry deletion = new TreeDeletionEntry(path, Long.MIN_VALUE, Long.MAX_VALUE); + FutureTask flush = + new FutureTask<>( + () -> { + processor.flushOneMemTable(); + return null; + }); + Thread flushThread = new Thread(flush, "test-flush-deletion-completion"); + Object flushingMemTables = processor.getFlushingMemTable(); + long monitorOwner = Thread.currentThread().getId(); + try { + synchronized (flushingMemTables) { + flushThread.start(); + // On the old path the flush has already scanned modifications when it reaches this + // monitor. The fixed path reaches it before scanning and releasing the memtable. + await() + .atMost(30, TimeUnit.SECONDS) + .until( + () -> { + ThreadInfo info = + ManagementFactory.getThreadMXBean().getThreadInfo(flushThread.getId()); + return info != null + && info.getThreadState() == Thread.State.BLOCKED + && info.getLockOwnerId() == monitorOwner + && info.getLockInfo().getIdentityHashCode() + == System.identityHashCode(flushingMemTables); + }); + DeleteDataNode node = + new DeleteDataNode( + new PlanNodeId(""), + Collections.singletonList(path), + Long.MIN_VALUE, + Long.MAX_VALUE); + assertEquals( + WALFlushListener.Status.SUCCESS, + processor.logDeleteDataNodeInWAL(node).waitForResult()); + assertTrue(processor.deleteDataInMemory(deletion)); + } + } finally { + try { + flush.get(30, TimeUnit.SECONDS); + } finally { + processor.setManagedByFlushManager(false); + } + } + assertEquals(0, processor.getFlushingMemTableSize()); + assertEquals( + Collections.singletonList(deletion), + ModificationFile.readAllModifications(new File(filePath), false)); + } finally { + try { + if (processor != null) { + processor.syncClose(); + } + } finally { + config.setWalMode(originalWalMode); + } + } + } + + @Test + public void testTreeDeletionWALWithoutWorkingMemTable() throws Exception { + testDeletionWALWithoutWorkingMemTable(false); + } + + @Test + public void testTableDeletionWALWithoutWorkingMemTable() throws Exception { + testDeletionWALWithoutWorkingMemTable(true); + } + + private void testDeletionWALWithoutWorkingMemTable(boolean tableModel) throws Exception { + WALMode originalWalMode = config.getWalMode(); + config.setWalMode(WALMode.SYNC); + try { + processor = + new TsFileProcessor( + storageGroup, + SystemFileFactory.INSTANCE.getFile(filePath), + sgInfo, + this::closeTsFileProcessor, + (tsFileProcessor, updateMap, systemFlushTime) -> {}, + true); + processor.setTsFileProcessorInfo(new TsFileProcessorInfo(sgInfo)); + sgInfo.initTsFileProcessorInfo(processor); + WALNode walNode = + (WALNode) + WALManager.getInstance() + .applyForWALNode(WALManager.getApplicantUniqueId(storageGroup, true)); + + Assert.assertNull(processor.getWorkMemTable()); + logDeletionForTest(tableModel, 1); + + TSRecord record = new TSRecord(deviceId, 1); + record.addTuple(DataPoint.getDataPoint(dataType, measurementId, "1")); + processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); + long memTableId = processor.getWorkMemTable().getMemTableId(); + logDeletionForTest(tableModel, 2); + + // Flush detaches the working memtable synchronously, while keeping the processor open. + long secondMemTableId; + processor.setManagedByFlushManager(true); + try { + processor.asyncFlush(); + Assert.assertNull(processor.getWorkMemTable()); + assertEquals(1, processor.getFlushingMemTableSize()); + processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); + secondMemTableId = processor.getWorkMemTable().getMemTableId(); + processor.asyncFlush(); + Assert.assertNull(processor.getWorkMemTable()); + assertEquals(2, processor.getFlushingMemTableSize()); + logDeletionForTest(tableModel, 3); + processor.flushOneMemTable(); + logDeletionForTest(tableModel, 4); + processor.flushOneMemTable(); + } finally { + processor.setManagedByFlushManager(false); + } + assertEquals(0, processor.getFlushingMemTableSize()); + logDeletionForTest(tableModel, 5); + walNode.rollWALFile(); + + List memTableIds = new ArrayList<>(); + List searchIndexes = new ArrayList<>(); + File[] walFiles = WALFileUtils.listAllWALFiles(walNode.getLogDirectory()); + Assert.assertNotNull(walFiles); + WALFileUtils.ascSortByVersionId(walFiles); + for (File walFile : walFiles) { + try (WALReader reader = new WALReader(walFile)) { + while (reader.hasNext()) { + WALEntry entry = reader.next(); + if (entry.getValue() instanceof DeleteDataNode) { + Assert.assertFalse(tableModel); + DeleteDataNode deletion = (DeleteDataNode) entry.getValue(); + assertEquals(Long.MIN_VALUE, deletion.getDeleteStartTime()); + assertEquals(Long.MAX_VALUE, deletion.getDeleteEndTime()); + assertEquals( + Collections.singletonList(new MeasurementPath(deviceId + "." + measurementId)), + deletion.getPathList()); + memTableIds.add(entry.getMemTableId()); + searchIndexes.add(deletion.getSearchIndex()); + } else if (entry.getValue() instanceof RelationalDeleteDataNode) { + Assert.assertTrue(tableModel); + RelationalDeleteDataNode deletion = (RelationalDeleteDataNode) entry.getValue(); + assertEquals( + Collections.singletonList( + new TableDeletionEntry( + new DeletionPredicate("t2"), + new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))), + deletion.getModEntries()); + memTableIds.add(entry.getMemTableId()); + searchIndexes.add(deletion.getSearchIndex()); + } + } + } + } + assertEquals( + Arrays.asList( + (long) TsFileProcessor.MEMTABLE_NOT_EXIST, + memTableId, + secondMemTableId, + secondMemTableId, + (long) TsFileProcessor.MEMTABLE_NOT_EXIST), + memTableIds); + assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L), searchIndexes); + } finally { + try { + if (processor != null) { + processor.syncClose(); + } + } finally { + config.setWalMode(originalWalMode); + } + } + } + + private void logDeletionForTest(boolean tableModel, long searchIndex) throws Exception { + WALFlushListener listener; + if (tableModel) { + RelationalDeleteDataNode deletion = + new RelationalDeleteDataNode( + new PlanNodeId(""), + new TableDeletionEntry( + new DeletionPredicate("t2"), new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE)), + storageGroup); + deletion.setSearchIndex(searchIndex); + listener = processor.logDeleteDataNodeInWAL(deletion); + } else { + DeleteDataNode deletion = + new DeleteDataNode( + new PlanNodeId(""), + Collections.singletonList(new MeasurementPath(deviceId + "." + measurementId)), + Long.MIN_VALUE, + Long.MAX_VALUE); + deletion.setSearchIndex(searchIndex); + listener = processor.logDeleteDataNodeInWAL(deletion); + } + assertEquals(WALFlushListener.Status.SUCCESS, listener.waitForResult()); + } + @Test public void testWriteAndFlush() throws IOException, WriteProcessException, MetadataException, ExecutionException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java index cecf04e6781d7..78950ed7492ee 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java @@ -24,11 +24,13 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; @@ -52,9 +54,13 @@ import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileReader; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.TsFileWriter; @@ -76,6 +82,8 @@ import java.nio.channels.FileChannel; import java.nio.file.Files; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; @@ -86,6 +94,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -158,6 +167,69 @@ public void testNormalProcedure() throws Exception { recoverAndCheck(); } + @Test + public void testRecoverMultipleMemTablesWithDeletionInSameTsFile() throws Exception { + IMemTable firstMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID); + IMemTable lastMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID); + // Both memtables are pending flush in the same file when the process crashes. + for (IMemTable memTable : Arrays.asList(firstMemTable, lastMemTable)) { + MemTableInfo info = + new MemTableInfo(memTable, FILE_WITH_WAL_NAME, walBuffer.getCurrentWALFileVersion()); + checkpointManager.makeCreateMemTableCPInMemory(info); + checkpointManager.makeCreateMemTableCPOnDisk(memTable.getMemTableId()); + } + checkpointManager.fsyncCheckpointFile(); + walBuffer.write( + new WALInfoEntry( + firstMemTable.getMemTableId(), getInsertRowNode(DEVICE2_NAME.toString(), 4L), true)); + walBuffer.write( + new WALInfoEntry( + lastMemTable.getMemTableId(), getInsertRowNode(DEVICE2_NAME.toString(), 5L), true)); + DeleteDataNode deletion = + new DeleteDataNode( + new PlanNodeId(""), + Collections.singletonList(new MeasurementPath(DEVICE2_NAME + ".s1")), + 4L, + 5L); + // TsFileProcessor associates deletion with the last pending memtable when workMemTable is null. + walBuffer.write(new WALInfoEntry(lastMemTable.getMemTableId(), deletion, true)); + walBuffer.close(); + bufferClosed = true; + + recoverManager.clear(); + File tsFile = new File(FILE_WITH_WAL_NAME); + generateCrashedFile(tsFile); + tsFileWithWALResource = new TsFileResource(tsFile); + UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer( + tsFileWithWALResource, true, p -> assertFalse(p.canWrite())); + recoverManager.addRecoverPerformer(performer); + recoverManager.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); + recoverManager.recover(); + assertEquals(WALRecoverListener.Status.SUCCESS, performer.getRecoverListener().waitForResult()); + + try (TsFileReader reader = new TsFileReader(new TsFileSequenceReader(FILE_WITH_WAL_NAME))) { + QueryDataSet dataSet = + reader.query( + QueryExpression.create( + Arrays.asList( + new Path(DEVICE2_NAME, "s1", true), new Path(DEVICE2_NAME, "s2", true)), + null)); + for (long time : new long[] {3L, 4L, 5L}) { + assertTrue(dataSet.hasNext()); + RowRecord record = dataSet.next(); + assertEquals(time, record.getTimestamp()); + if (time == 3L) { + assertEquals(3.0f, record.getFields().get(0).getFloatV(), 0); + } else { + assertNull(record.getFields().get(0)); + } + assertEquals(time == 3L ? 3.0d : 1.0d, record.getFields().get(1).getDoubleV(), 0); + } + assertFalse(dataSet.hasNext()); + } + } + private void prepareCheckpointAndWALFileForNormal() throws MetadataException, ExecutionException, InterruptedException, QueryProcessException { // write useless .wal files, start write threads to write concurrently From 08a98a610530e9436be54e2fb03db6fee37116a8 Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Thu, 15 Oct 2026 10:08:59 +0800 Subject: [PATCH 2/2] Keep deletion fixes within single-memtable file lifecycle --- .../dataregion/memtable/TsFileProcessor.java | 33 ++------- .../wal/recover/WALNodeRecoverTask.java | 16 ++--- .../memtable/TsFileProcessorTest.java | 35 ++------- .../wal/recover/WALRecoverManagerTest.java | 72 ------------------- 4 files changed, 17 insertions(+), 139 deletions(-) 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 fda5bad3bb94d..f6157e5e262d0 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 @@ -1625,36 +1625,17 @@ public boolean deleteDataInMemory(ModEntry deletion) { } } + // A processor may exist before its first memtable is created if table registration fails. public WALFlushListener logDeleteDataNodeInWAL(DeleteDataNode deleteDataNode) { - flushQueryLock.readLock().lock(); - try { - walEntryNum++; - return walNode.log(getMemTableIdForDeletion(), deleteDataNode); - } finally { - flushQueryLock.readLock().unlock(); - } + walEntryNum++; + return walNode.log( + workMemTable == null ? MEMTABLE_NOT_EXIST : workMemTable.getMemTableId(), deleteDataNode); } public WALFlushListener logDeleteDataNodeInWAL(RelationalDeleteDataNode deleteDataNode) { - flushQueryLock.readLock().lock(); - try { - walEntryNum++; - return walNode.log(getMemTableIdForDeletion(), deleteDataNode); - } finally { - flushQueryLock.readLock().unlock(); - } - } - - // Called with flushQueryLock held until the deletion is queued in WAL, before the selected - // memtable can be released and its flush checkpoint recorded. - private long getMemTableIdForDeletion() { - if (workMemTable != null) { - return workMemTable.getMemTableId(); - } - IMemTable flushingMemTable = flushingMemTables.peekLast(); - // A pending flush still needs this deletion during recovery. If no memtable remains, keep - // logging with the existing sentinel so that IoTConsensus can replicate the deletion. - return flushingMemTable == null ? MEMTABLE_NOT_EXIST : flushingMemTable.getMemTableId(); + walEntryNum++; + return walNode.log( + workMemTable == null ? MEMTABLE_NOT_EXIST : workMemTable.getMemTableId(), deleteDataNode); } public TsFileResource getTsFileResource() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java index 30fdf97c414bc..7efce7036563b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java @@ -54,7 +54,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; -import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicLong; @@ -74,8 +73,6 @@ public class WALNodeRecoverTask implements Runnable { private long firstValidVersionId = Long.MAX_VALUE; private Map memTableId2Info; private Map memTableId2RecoverPerformer; - // Multiple unflushed memtables in one TsFile share a single recovery lifecycle. - private final Set recoverPerformers = new HashSet<>(); public WALNodeRecoverTask(File logDirectory, CountDownLatch allNodesRecoveredLatch) { this.logDirectory = logDirectory; @@ -95,11 +92,11 @@ public void run() { recoverInfoFromCheckpoints(); recoverTsFiles(); } catch (Exception e) { - for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { recoverPerformer.getRecoverListener().fail(e); } } finally { - for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { try { if (!recoverPerformer.canWrite()) { recoverPerformer.close(); @@ -266,16 +263,13 @@ private void recoverInfoFromCheckpoints() { } // update firstValidVersionId and get recover performer from WALRecoverManager memTableId2RecoverPerformer = new HashMap<>(); - Map file2RecoverPerformer = new HashMap<>(); for (MemTableInfo memTableInfo : memTableId2Info.values()) { firstValidVersionId = Math.min(firstValidVersionId, memTableInfo.getFirstFileVersionId()); - File tsFile = new File(memTableInfo.getTsFilePath()); UnsealedTsFileRecoverPerformer recoverPerformer = - file2RecoverPerformer.computeIfAbsent(tsFile, walRecoverManger::removeRecoverPerformer); + walRecoverManger.removeRecoverPerformer(new File(memTableInfo.getTsFilePath())); if (recoverPerformer != null) { memTableId2RecoverPerformer.put(memTableInfo.getMemTableId(), recoverPerformer); - recoverPerformers.add(recoverPerformer); } } } @@ -285,7 +279,7 @@ private void recoverTsFiles() { return; } // make preparation for recovery - for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { try { recoverPerformer.startRecovery(); } catch (Exception e) { @@ -354,7 +348,7 @@ private void recoverTsFiles() { private void endRecovery() { // end recovering all recover performers - for (UnsealedTsFileRecoverPerformer recoverPerformer : recoverPerformers) { + for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { try { recoverPerformer.endRecovery(); recoverPerformer.getRecoverListener().succeed(); 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 29c8bed8b0efd..324bf78a37e09 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 @@ -237,7 +237,7 @@ public void testDeletionDuringFlushCompletionIsPersisted() throws Exception { record.addTuple(DataPoint.getDataPoint(dataType, measurementId, "1")); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); processor.setManagedByFlushManager(true); - processor.asyncFlush(); + processor.asyncClose(); Assert.assertNull(processor.getWorkMemTable()); MeasurementPath path = new MeasurementPath(deviceId + "." + measurementId); @@ -268,15 +268,6 @@ public void testDeletionDuringFlushCompletionIsPersisted() throws Exception { && info.getLockInfo().getIdentityHashCode() == System.identityHashCode(flushingMemTables); }); - DeleteDataNode node = - new DeleteDataNode( - new PlanNodeId(""), - Collections.singletonList(path), - Long.MIN_VALUE, - Long.MAX_VALUE); - assertEquals( - WALFlushListener.Status.SUCCESS, - processor.logDeleteDataNodeInWAL(node).waitForResult()); assertTrue(processor.deleteDataInMemory(deletion)); } } finally { @@ -339,27 +330,17 @@ private void testDeletionWALWithoutWorkingMemTable(boolean tableModel) throws Ex long memTableId = processor.getWorkMemTable().getMemTableId(); logDeletionForTest(tableModel, 2); - // Flush detaches the working memtable synchronously, while keeping the processor open. - long secondMemTableId; + // Use the production close path, while keeping flush execution under test control. processor.setManagedByFlushManager(true); try { - processor.asyncFlush(); + processor.asyncClose(); Assert.assertNull(processor.getWorkMemTable()); assertEquals(1, processor.getFlushingMemTableSize()); - processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); - secondMemTableId = processor.getWorkMemTable().getMemTableId(); - processor.asyncFlush(); - Assert.assertNull(processor.getWorkMemTable()); - assertEquals(2, processor.getFlushingMemTableSize()); - logDeletionForTest(tableModel, 3); - processor.flushOneMemTable(); - logDeletionForTest(tableModel, 4); processor.flushOneMemTable(); } finally { processor.setManagedByFlushManager(false); } assertEquals(0, processor.getFlushingMemTableSize()); - logDeletionForTest(tableModel, 5); walNode.rollWALFile(); List memTableIds = new ArrayList<>(); @@ -397,14 +378,8 @@ private void testDeletionWALWithoutWorkingMemTable(boolean tableModel) throws Ex } } assertEquals( - Arrays.asList( - (long) TsFileProcessor.MEMTABLE_NOT_EXIST, - memTableId, - secondMemTableId, - secondMemTableId, - (long) TsFileProcessor.MEMTABLE_NOT_EXIST), - memTableIds); - assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L), searchIndexes); + Arrays.asList((long) TsFileProcessor.MEMTABLE_NOT_EXIST, memTableId), memTableIds); + assertEquals(Arrays.asList(1L, 2L), searchIndexes); } finally { try { if (processor != null) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java index 78950ed7492ee..cecf04e6781d7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java @@ -24,13 +24,11 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; -import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; @@ -54,13 +52,9 @@ import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.enums.TSEncoding; -import org.apache.tsfile.read.TsFileReader; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.Chunk; import org.apache.tsfile.read.common.Path; -import org.apache.tsfile.read.common.RowRecord; -import org.apache.tsfile.read.expression.QueryExpression; -import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.TsFileWriter; @@ -82,8 +76,6 @@ import java.nio.channels.FileChannel; import java.nio.file.Files; import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; @@ -94,7 +86,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -167,69 +158,6 @@ public void testNormalProcedure() throws Exception { recoverAndCheck(); } - @Test - public void testRecoverMultipleMemTablesWithDeletionInSameTsFile() throws Exception { - IMemTable firstMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID); - IMemTable lastMemTable = new PrimitiveMemTable(SG_NAME, DATA_REGION_ID); - // Both memtables are pending flush in the same file when the process crashes. - for (IMemTable memTable : Arrays.asList(firstMemTable, lastMemTable)) { - MemTableInfo info = - new MemTableInfo(memTable, FILE_WITH_WAL_NAME, walBuffer.getCurrentWALFileVersion()); - checkpointManager.makeCreateMemTableCPInMemory(info); - checkpointManager.makeCreateMemTableCPOnDisk(memTable.getMemTableId()); - } - checkpointManager.fsyncCheckpointFile(); - walBuffer.write( - new WALInfoEntry( - firstMemTable.getMemTableId(), getInsertRowNode(DEVICE2_NAME.toString(), 4L), true)); - walBuffer.write( - new WALInfoEntry( - lastMemTable.getMemTableId(), getInsertRowNode(DEVICE2_NAME.toString(), 5L), true)); - DeleteDataNode deletion = - new DeleteDataNode( - new PlanNodeId(""), - Collections.singletonList(new MeasurementPath(DEVICE2_NAME + ".s1")), - 4L, - 5L); - // TsFileProcessor associates deletion with the last pending memtable when workMemTable is null. - walBuffer.write(new WALInfoEntry(lastMemTable.getMemTableId(), deletion, true)); - walBuffer.close(); - bufferClosed = true; - - recoverManager.clear(); - File tsFile = new File(FILE_WITH_WAL_NAME); - generateCrashedFile(tsFile); - tsFileWithWALResource = new TsFileResource(tsFile); - UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer( - tsFileWithWALResource, true, p -> assertFalse(p.canWrite())); - recoverManager.addRecoverPerformer(performer); - recoverManager.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); - recoverManager.recover(); - assertEquals(WALRecoverListener.Status.SUCCESS, performer.getRecoverListener().waitForResult()); - - try (TsFileReader reader = new TsFileReader(new TsFileSequenceReader(FILE_WITH_WAL_NAME))) { - QueryDataSet dataSet = - reader.query( - QueryExpression.create( - Arrays.asList( - new Path(DEVICE2_NAME, "s1", true), new Path(DEVICE2_NAME, "s2", true)), - null)); - for (long time : new long[] {3L, 4L, 5L}) { - assertTrue(dataSet.hasNext()); - RowRecord record = dataSet.next(); - assertEquals(time, record.getTimestamp()); - if (time == 3L) { - assertEquals(3.0f, record.getFields().get(0).getFloatV(), 0); - } else { - assertNull(record.getFields().get(0)); - } - assertEquals(time == 3L ? 3.0d : 1.0d, record.getFields().get(1).getDoubleV(), 0); - } - assertFalse(dataSet.hasNext()); - } - } - private void prepareCheckpointAndWALFileForNormal() throws MetadataException, ExecutionException, InterruptedException, QueryProcessException { // write useless .wal files, start write threads to write concurrently