This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/fix-page-cache-deletion-dal in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d27d37b0a214fa934e2991b0a244262a59f04f42 Author: Caideyipi <[email protected]> AuthorDate: Mon Oct 12 18:02:08 2026 +0800 [Pipe] Fix DAL persistence acknowledgments --- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 3 + .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 3 + .../deletion/persist/PageCacheDeletionBuffer.java | 17 +- .../persist/PageCacheDeletionBufferTest.java | 196 +++++++++++++++++++++ 4 files changed, 215 insertions(+), 4 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 836ac88d84f..95cccd1efdd 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -66,6 +66,9 @@ public final class DataNodePipeMessages { // ===================== CONSENSUS ===================== + public static final String EXCEPTION_DELETION_RESOURCE_EXCEEDS_THE_DAL_BUFFER_SIZE_E2A17CAC = + "Deletion resource exceeds the DAL buffer size"; + public static final String CLOSING_DELETION_RESOURCE_MANAGER_FOR = "Closing deletion resource manager for {}..."; public static final String DAL_THREAD_STILL_DOESN_T_EXIT_AFTER = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 5840da12daf..47e9e351241 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -66,6 +66,9 @@ public final class DataNodePipeMessages { // ===================== CONSENSUS ===================== + public static final String EXCEPTION_DELETION_RESOURCE_EXCEEDS_THE_DAL_BUFFER_SIZE_E2A17CAC = + "删除资源超过 DAL 缓冲区大小"; + public static final String CLOSING_DELETION_RESOURCE_MANAGER_FOR = "正在关闭 {} 的删除资源管理器..."; public static final String DAL_THREAD_STILL_DOESN_T_EXIT_AFTER = "DAL 线程 {} 在 30 秒后仍未退出"; public static final String DELETIONMANAGER_CURRENT_DAL_DIR_IS_DELETED_SUCCESSFULLY = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java index f306ee15eee..6fe2246627f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java @@ -195,8 +195,11 @@ public class PageCacheDeletionBuffer implements DeletionBuffer { private void closeCurrentLoggingFile(boolean notifySuccess) throws IOException { LOGGER.info(DataNodePipeMessages.DELETION_PERSIST_CURRENT_FILE_HAS_BEEN_CLOSED, dataRegionId); - // Close old resource to fsync. if (this.logChannel != null) { + // Closing a channel does not fsync. Persist the rotated batch before acknowledging it. + if (notifySuccess) { + this.logChannel.force(false); + } this.logChannel.close(); } if (notifySuccess) { @@ -336,6 +339,8 @@ public class PageCacheDeletionBuffer implements DeletionBuffer { // if any exception occurred, this batch will not be written to disk and lost. pendingDeletionsInOneTask.forEach(deletionResource -> deletionResource.onPersistFailed(e)); rollbackFileAttribute(currentTaskBatchSize.get()); + // Failed deletions must not remain in the next batch's success callbacks. + resetTaskAttribute(); taskFinished = true; } finally { if (taskFinished) { @@ -372,11 +377,15 @@ public class PageCacheDeletionBuffer implements DeletionBuffer { // For first deletion we use blocking take() method. try { DeletionResource firstDeletionResource = deletionResources.take(); - // Serialize deletion. The first serialization cannot fail because a deletion cannot exceed - // size of serializeBuffer. - serializeDeletionToBatchBuffer(firstDeletionResource); pendingDeletionsInOneTask.add(firstDeletionResource); currentTaskDeletionCount++; + // A deletion can exceed the DAL buffer even when it is the first entry in a batch. + // Include it in failure handling so it cannot be acknowledged without being written. + if (!serializeDeletionToBatchBuffer(firstDeletionResource)) { + throw new IOException( + DataNodePipeMessages + .EXCEPTION_DELETION_RESOURCE_EXCEEDS_THE_DAL_BUFFER_SIZE_E2A17CAC); + } maxProgressIndexInCurrentFile = maxProgressIndexInCurrentFile.updateToMinimumEqualOrIsAfterProgressIndex( firstDeletionResource.getProgressIndex()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java index b63dff6a5b6..1b2c0b89245 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java @@ -25,21 +25,55 @@ import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource; +import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager; +import org.apache.iotdb.db.pipe.consensus.deletion.recover.DeletionReader; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; +import org.awaitility.Awaitility; +import org.junit.After; import org.junit.Assert; +import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.mockito.InOrder; +import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.util.Collections; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +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.withSettings; public class PageCacheDeletionBufferTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + private int previousDalBufferSize; + + @Before + public void setUp() { + previousDalBufferSize = PageCacheDeletionBuffer.DAL_BUFFER_SIZE; + } + + @After + public void tearDown() { + PageCacheDeletionBuffer.setDalBufferSize(previousDalBufferSize); + } @Test public void testDequeuedDeletionIsNotReportedAsFlushed() throws Exception { @@ -85,4 +119,166 @@ public class PageCacheDeletionBufferTest { } Assert.assertSame(DeletionResource.Status.SUCCESS, deletionResource.waitForResult()); } + + @Test(timeout = 30000) + public void testOversizedFirstDeletionFailsWithoutAffectingNextDeletion() throws Exception { + final DeletionResource oversizedDeletion = createDeletion(1, 32); + final DeletionResource nextDeletion = createDeletion(2, 1); + PageCacheDeletionBuffer.setDalBufferSize(1024); + Assert.assertTrue( + oversizedDeletion.serialize().position() > PageCacheDeletionBuffer.DAL_BUFFER_SIZE); + Assert.assertTrue( + nextDeletion.serialize().position() < PageCacheDeletionBuffer.DAL_BUFFER_SIZE); + final File directory = temporaryFolder.newFolder("oversized"); + final PageCacheDeletionBuffer deletionBuffer = + new PageCacheDeletionBuffer(1, directory.getAbsolutePath()); + deletionBuffer.start(); + try { + deletionBuffer.registerDeletionResource(oversizedDeletion); + Assert.assertSame(DeletionResource.Status.FAILURE, oversizedDeletion.waitForResult()); + Assert.assertTrue(oversizedDeletion.getCause() instanceof IOException); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(deletionBuffer::isAllDeletionFlushed); + + deletionBuffer.registerDeletionResource(nextDeletion); + Assert.assertSame(DeletionResource.Status.SUCCESS, nextDeletion.waitForResult()); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(deletionBuffer::isAllDeletionFlushed); + Assert.assertSame(DeletionResource.Status.FAILURE, oversizedDeletion.waitForResult()); + } finally { + deletionBuffer.close(); + } + try (DeletionReader reader = + new DeletionReader( + new File(directory, "_0-0" + DeletionResourceManager.DELETION_FILE_SUFFIX), + 1, + ignored -> {})) { + final List<DeletionResource> recovered = reader.readAllDeletions(); + Assert.assertEquals(1, recovered.size()); + Assert.assertEquals(nextDeletion.getDeleteDataNode(), recovered.get(0).getDeleteDataNode()); + } + } + + @Test(timeout = 30000) + public void testThresholdRotationForcesBeforeSuccess() throws Exception { + assertRotationForcesBeforeSuccess(true); + } + + @Test(timeout = 30000) + public void testFullBatchRotationForcesBeforeSuccess() throws Exception { + assertRotationForcesBeforeSuccess(false); + } + + private void assertRotationForcesBeforeSuccess(boolean reachesThreshold) throws Exception { + final DeletionResource firstDeletion = spy(createDeletion(1, 1)); + final DeletionResource nextDeletion = createDeletion(2, 2); + final int firstDeletionSize = firstDeletion.serialize().position(); + PageCacheDeletionBuffer.setDalBufferSize( + reachesThreshold ? firstDeletionSize : firstDeletionSize * 2); + final FileChannel channel = createLoggingChannel(); + final PageCacheDeletionBuffer deletionBuffer = + new PageCacheDeletionBuffer(1, temporaryFolder.newFolder("rotation").getAbsolutePath()); + deletionBuffer.start(); + replaceLoggingChannel(deletionBuffer, channel); + final AtomicBoolean closedBeforeSuccess = new AtomicBoolean(); + final CountDownLatch deletionsRegistered = new CountDownLatch(1); + doAnswer( + invocation -> { + Assert.assertTrue(deletionsRegistered.await(10, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }) + .when(firstDeletion) + .serialize(); + doAnswer( + invocation -> { + closedBeforeSuccess.set(!channel.isOpen()); + invocation.callRealMethod(); + return null; + }) + .when(firstDeletion) + .onPersistSucceed(); + try { + deletionBuffer.registerDeletionResource(firstDeletion); + if (!reachesThreshold) { + deletionBuffer.registerDeletionResource(nextDeletion); + } + deletionsRegistered.countDown(); + Assert.assertSame(DeletionResource.Status.SUCCESS, firstDeletion.waitForResult()); + Assert.assertTrue(closedBeforeSuccess.get()); + if (!reachesThreshold) { + Assert.assertSame(DeletionResource.Status.SUCCESS, nextDeletion.waitForResult()); + } + final InOrder order = inOrder(channel, firstDeletion); + order.verify(channel).write(any(ByteBuffer.class)); + order.verify(channel).force(false); + order.verify(firstDeletion).onPersistSucceed(); + } finally { + deletionsRegistered.countDown(); + deletionBuffer.close(); + } + } + + @Test(timeout = 30000) + public void testRotationForceFailureDoesNotReportSuccess() throws Exception { + final DeletionResource firstDeletion = spy(createDeletion(1, 1)); + final DeletionResource nextDeletion = createDeletion(2, 1); + PageCacheDeletionBuffer.setDalBufferSize(firstDeletion.serialize().position()); + final FileChannel channel = createLoggingChannel(); + final IOException failure = new IOException("injected force failure"); + doThrow(failure).when(channel).force(false); + final PageCacheDeletionBuffer deletionBuffer = + new PageCacheDeletionBuffer( + 1, temporaryFolder.newFolder("force-failure").getAbsolutePath()); + deletionBuffer.start(); + replaceLoggingChannel(deletionBuffer, channel); + try { + deletionBuffer.registerDeletionResource(firstDeletion); + Assert.assertSame(DeletionResource.Status.FAILURE, firstDeletion.waitForResult()); + Assert.assertSame(failure, firstDeletion.getCause()); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(deletionBuffer::isAllDeletionFlushed); + + doNothing().when(channel).force(false); + deletionBuffer.registerDeletionResource(nextDeletion); + Assert.assertSame(DeletionResource.Status.SUCCESS, nextDeletion.waitForResult()); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(deletionBuffer::isAllDeletionFlushed); + Assert.assertSame(DeletionResource.Status.FAILURE, firstDeletion.waitForResult()); + verify(firstDeletion, never()).onPersistSucceed(); + } finally { + deletionBuffer.close(); + } + } + + private DeletionResource createDeletion(long progressIndex, int pathCount) throws Exception { + final DeleteDataNode node = + new DeleteDataNode( + new PlanNodeId("1"), + Collections.nCopies(pathCount, new MeasurementPath("root.vehicle.d2.s0")), + 50, + 150); + node.setProgressIndex( + new RecoverProgressIndex( + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), + new SimpleProgressIndex(0, progressIndex))); + return new DeletionResource(node, ignored -> {}, 1); + } + + private FileChannel createLoggingChannel() throws IOException { + final FileChannel channel = mock(FileChannel.class, withSettings().useConstructor()); + doAnswer( + invocation -> { + final ByteBuffer buffer = invocation.getArgument(0); + final int size = buffer.remaining(); + buffer.position(buffer.limit()); + return size; + }) + .when(channel) + .write(any(ByteBuffer.class)); + return channel; + } + + private void replaceLoggingChannel(PageCacheDeletionBuffer deletionBuffer, FileChannel channel) + throws Exception { + final Field field = PageCacheDeletionBuffer.class.getDeclaredField("logChannel"); + field.setAccessible(true); + ((FileChannel) field.get(deletionBuffer)).close(); + field.set(deletionBuffer, channel); + } }
