Caideyipi · Oct 12, 2026
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
@@ -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
@@ -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/...ain/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java b/...ain/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
@@ -195,8 +195,11 @@ private void fsyncCurrentLoggingFile() throws IOException {
 
   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 void run() {
         // 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 @@ private void persistDeletion() throws IOException {
       // 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/...java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java b/...java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBufferTest.java
@@ -25,21 +25,55 @@
 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 ByteBuffer serialize() {
     }
     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);
+  }
 }