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..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,14 +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) { walEntryNum++; - return walNode.log(workMemTable.getMemTableId(), deleteDataNode); + return walNode.log( + workMemTable == null ? MEMTABLE_NOT_EXIST : workMemTable.getMemTableId(), deleteDataNode); } public WALFlushListener logDeleteDataNodeInWAL(RelationalDeleteDataNode deleteDataNode) { walEntryNum++; - return walNode.log(workMemTable.getMemTableId(), deleteDataNode); + return walNode.log( + workMemTable == null ? MEMTABLE_NOT_EXIST : workMemTable.getMemTableId(), deleteDataNode); } public TsFileResource getTsFileResource() { @@ -1983,39 +1986,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/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..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 @@ -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,203 @@ 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.asyncClose(); + 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); + }); + 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); + + // Use the production close path, while keeping flush execution under test control. + processor.setManagedByFlushManager(true); + try { + processor.asyncClose(); + Assert.assertNull(processor.getWorkMemTable()); + assertEquals(1, processor.getFlushingMemTableSize()); + processor.flushOneMemTable(); + } finally { + processor.setManagedByFlushManager(false); + } + assertEquals(0, processor.getFlushingMemTableSize()); + 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), memTableIds); + assertEquals(Arrays.asList(1L, 2L), 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 {