From 7c5abb3d460df8f8a8fc8013f3d944a71e6f1b62 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:17:51 +0800 Subject: [PATCH] [To dev/1.3] [Pipe] Preserve historical TsFile mods after compaction (#18904) Backport only the historical TsFile/mod reference fix. Adapt to the 1.3 ModificationFile API and use the same creation-time-qualified pipe directory for historical selection and events. (cherry picked from commit 10c5f89ba52ec39cd6bfb6b8b735287cc0d935f9) --- .../tsfile/PipeTsFileInsertionEvent.java | 18 +++- .../tsfile/PipeTsFileResourceManager.java | 38 +++++--- .../PipeHistoricalDataRegionTsFileSource.java | 22 ++++- .../event/PipeTsFileInsertionEventTest.java | 97 +++++++++++++++++++ .../PipeTsFileResourceManagerTest.java | 48 +++++++++ ...eHistoricalDataRegionTsFileSourceTest.java | 68 +++++++++++++ 6 files changed, 270 insertions(+), 21 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java index 0d379db7831b5..ae0547529b6ed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java @@ -189,7 +189,9 @@ private PipeTsFileInsertionEvent( this.tsFile = Objects.isNull(tsFile) ? resource.getTsFile() : tsFile; final ModificationFile modFile = resource.getModFile(); - this.isWithMod = isWithMod && modFile.exists(); + // Historical files and their mods have already been pinned. Compaction may have removed + // the original resource before this event is created, so inspect the pinned mods later. + this.isWithMod = isWithMod && (isGeneratedByHistoricalExtractor || modFile.exists()); this.modFile = this.isWithMod ? new File(modFile.getFilePath()) : null; this.isLoaded = isLoaded; @@ -364,9 +366,22 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); final File originalTsFile = tsFile; final File originalModFile = modFile; + final boolean originalIsWithMod = isWithMod; File increasedTsFile = null; boolean increased = false; try { + if (isGeneratedByHistoricalExtractor) { + tsFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + tsFile, pipeTsFileResourcePipeName); + if (isWithMod) { + modFile = new File(tsFile + ModificationFile.FILE_SUFFIX); + isWithMod = modFile.exists(); + if (!isWithMod) { + modFile = null; + } + } + } increasedTsFile = PipeDataNodeResourceManager.tsfile() .increaseFileReference(tsFile, true, pipeTsFileResourcePipeName); @@ -389,6 +404,7 @@ public boolean internallyIncreaseResourceReferenceCount(final String holderMessa } tsFile = originalTsFile; modFile = originalModFile; + isWithMod = originalIsWithMod; LOGGER.warn( String.format( "Increase reference count for TsFile %s or modFile %s error. Holder Message: %s", diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java index 0dfefda76e499..542d610d6fe21 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java @@ -427,7 +427,8 @@ public Map getMeasurementDataTypeMapFromCache( : hardlinkOrCopiedFileToTsFilePublicResourceMap; } - public void pinTsFileResource( + /** Returns whether a companion mod-file reference was acquired. */ + public boolean pinTsFileResource( final TsFileResource resource, final boolean withMods, final @Nullable String pipeName) throws IOException { final File pinnedTsFile = increaseFileReference(resource.getTsFile(), true, pipeName); @@ -438,7 +439,9 @@ public void pinTsFileResource( synchronized (modFile) { increaseFileReference(new File(modFile.getFilePath()), false, pipeName); } + return true; } + return false; } catch (final IOException | RuntimeException e) { // Pinning is a two-file operation. Do not leave the TsFile pinned when the mod file cannot // be pinned (for example, when the pipe directory is temporarily unavailable). @@ -451,7 +454,8 @@ public void pinTsFileResource( } } - public void unpinTsFileResource(final TsFileResource resource, final @Nullable String pipeName) + public void unpinTsFileResource( + final TsFileResource resource, final @Nullable String pipeName, final boolean pinnedModFile) throws IOException { Exception firstException = null; File pinnedFile = null; @@ -464,20 +468,22 @@ public void unpinTsFileResource(final TsFileResource resource, final @Nullable S // Always attempt the mod-file cleanup even when resolving/decreasing the TsFile fails. A // failed first cleanup must not strand the second reference. - try { - final File pinnedModFile = - pinnedFile != null - ? new File(pinnedFile + ModificationFile.FILE_SUFFIX) - : getHardlinkOrCopiedFileInPipeDir( - new File(resource.getModFile().getFilePath()), pipeName); - if (pinnedModFile.exists()) { - decreaseFileReference(pinnedModFile, pipeName); - } - } catch (final IOException | RuntimeException e) { - if (firstException == null) { - firstException = e; - } else { - firstException.addSuppressed(e); + // Use the state captured when pinning. Compaction can remove the original mod file, and + // an event can acquire mod-file references even when no mod file was originally pinned. + if (pinnedModFile) { + try { + decreaseFileReference( + pinnedFile != null + ? new File(pinnedFile + ModificationFile.FILE_SUFFIX) + : getHardlinkOrCopiedFileInPipeDir( + new File(resource.getModFile().getFilePath()), pipeName), + pipeName); + } catch (final IOException | RuntimeException e) { + if (firstException == null) { + firstException = e; + } else { + firstException.addSuppressed(e); + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java index 1d984f9e18f45..9814bb9259c1e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java @@ -38,6 +38,7 @@ import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; @@ -109,6 +110,7 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR private String pipeName; private long creationTime; + private String pipeNameWithCreationTime; private PipeTaskMeta pipeTaskMeta; private ProgressIndex startIndex; @@ -141,6 +143,7 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR private Queue pendingQueue; private final Set filteredTsFileResources = new HashSet<>(); + private final Set resourcesWithPinnedModFile = new HashSet<>(); private final Set historicalProgressReportResources = new HashSet<>(); private ProgressIndex maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE; private ProgressIndex maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE; @@ -293,6 +296,8 @@ public void customize( pipeName = environment.getPipeName(); creationTime = environment.getCreationTime(); + pipeNameWithCreationTime = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); pipeTaskMeta = environment.getPipeTaskMeta(); startIndex = environment.getPipeTaskMeta().getProgressIndex(); @@ -424,8 +429,10 @@ public synchronized void start() { // Pin the resource, in case the file is removed by compaction or anything. // Will unpin it after the PipeTsFileInsertionEvent is created and pinned. try { - PipeDataNodeResourceManager.tsfile() - .pinTsFileResource(resource, shouldTransferModFile, pipeName); + if (PipeDataNodeResourceManager.tsfile() + .pinTsFileResource(resource, shouldTransferModFile, pipeNameWithCreationTime)) { + resourcesWithPinnedModFile.add(resource); + } return false; } catch (final IOException e) { LOGGER.warn("Pipe: failed to pin TsFileResource {}", resource.getTsFilePath(), e); @@ -886,7 +893,9 @@ private Event supplyTsFileEvent(final TsFileResource resource) { return isReferenceCountIncreased ? event : null; } finally { try { - PipeDataNodeResourceManager.tsfile().unpinTsFileResource(resource, pipeName); + PipeDataNodeResourceManager.tsfile() + .unpinTsFileResource( + resource, pipeNameWithCreationTime, resourcesWithPinnedModFile.remove(resource)); } catch (final IOException e) { LOGGER.warn( "Pipe {}@{}: failed to unpin TsFileResource after creating event, original path: {}", @@ -919,7 +928,11 @@ public synchronized void close() { pendingQueue.forEach( resource -> { try { - PipeDataNodeResourceManager.tsfile().unpinTsFileResource(resource, pipeName); + PipeDataNodeResourceManager.tsfile() + .unpinTsFileResource( + resource, + pipeNameWithCreationTime, + resourcesWithPinnedModFile.remove(resource)); } catch (final IOException e) { LOGGER.warn( "Pipe {}@{}: failed to unpin TsFileResource after dropping pipe, original path: {}", @@ -931,6 +944,7 @@ public synchronized void close() { pendingQueue.clear(); pendingQueue = null; } + resourcesWithPinnedModFile.clear(); historicalProgressReportResources.clear(); pendingHistoricalProgressIndexToReport = null; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java index b30cf4f98e3f5..c8098e498105f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java @@ -20,9 +20,15 @@ package org.apache.iotdb.db.pipe.event; import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; +import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.DeviceTimeIndex; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex; @@ -34,9 +40,100 @@ import java.io.File; import java.io.IOException; import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Collections; public class PipeTsFileInsertionEventTest { + @Test + public void testHistoricalModsSurviveOriginalFileRemoval() throws Exception { + assertHistoricalModsAfterOriginalFileRemoval(true, true); + } + + @Test + public void testHistoricalFileWithoutModsSurvivesOriginalFileRemoval() throws Exception { + assertHistoricalModsAfterOriginalFileRemoval(false, true); + } + + @Test + public void testHistoricalModsRemainDisabledAfterOriginalFileRemoval() throws Exception { + assertHistoricalModsAfterOriginalFileRemoval(true, false); + } + + private void assertHistoricalModsAfterOriginalFileRemoval( + final boolean hasMods, final boolean transferMods) throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalMods").toFile(); + final String pipeName = "historical_mods"; + final long creationTime = 1L; + final String resourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); + final PipeTsFileResourceManager manager = PipeDataNodeResourceManager.tsfile(); + PipeTsFileInsertionEvent event = null; + boolean increased = false; + File pinnedTsFile = null; + File pinnedModFile = null; + try { + final File dataDir = new File(tempDir, "unsequence/root.db/0/0"); + Assert.assertTrue(dataDir.mkdirs()); + final TsFileResource resource = + createNonEmptyTsFileResource(dataDir, "1-1-0-0.tsfile", 1L, 0); + resource.setStatusForTest(TsFileResourceStatus.NORMAL); + final Deletion deletion = new Deletion(new PartialPath("root.db.d0.s1"), 0, 1L, 1L); + if (hasMods) { + resource.getModFile().write(deletion); + resource.getModFile().close(); + } + manager.pinTsFileResource(resource, transferMods, resourcePipeName); + pinnedTsFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + resource.getTsFile(), resourcePipeName); + pinnedModFile = new File(pinnedTsFile + ModificationFile.FILE_SUFFIX); + + // Compaction removes the original after historical selection has pinned both files. + Assert.assertTrue(resource.remove()); + Assert.assertFalse(resource.modFileExists()); + Assert.assertEquals(hasMods && transferMods, pinnedModFile.exists()); + event = + new PipeTsFileInsertionEvent( + resource, + null, + transferMods, + false, + true, + pipeName, + creationTime, + null, + null, + Long.MIN_VALUE, + Long.MAX_VALUE); + increased = event.internallyIncreaseResourceReferenceCount(getClass().getName()); + Assert.assertTrue(increased); + Assert.assertEquals(hasMods && transferMods, event.isWithMod()); + if (hasMods && transferMods) { + Assert.assertEquals(pinnedModFile, event.getModFile()); + try (final ModificationFile mods = new ModificationFile(event.getModFile().getPath())) { + Assert.assertEquals( + Collections.singletonList(deletion), new ArrayList<>(mods.getModifications())); + } + } else { + Assert.assertNull(event.getModFile()); + } + } finally { + if (increased) { + event.internallyDecreaseResourceReferenceCount(getClass().getName()); + } else if (event != null) { + event.close(); + } + if (pinnedModFile != null) { + manager.decreaseFileReference(pinnedModFile, resourcePipeName); + } + if (pinnedTsFile != null) { + manager.decreaseFileReference(pinnedTsFile, resourcePipeName); + } + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test(timeout = 60000) public void testRealtimeEventCanSkipWaitingForClosedStatusAfterTsFileSealed() throws Exception { final File tempDir = Files.createTempDirectory("pipeTsFileSealed").toFile(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/PipeTsFileResourceManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/PipeTsFileResourceManagerTest.java index 06f11ed9b763f..07b245b233ee1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/PipeTsFileResourceManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/PipeTsFileResourceManagerTest.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; import org.apache.iotdb.db.storageengine.dataregion.modification.Modification; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.enums.TSEncoding; @@ -239,6 +240,53 @@ public void testDecreaseTsFile() throws IOException { Assert.assertFalse(Files.exists(originModFile.toPath())); } + @Test + public void testUnpinAfterOriginalModsRemoved() throws Exception { + final TsFileResource resource = new TsFileResource(new File(TS_FILE_NAME)); + final boolean pinnedModFile = + pipeTsFileResourceManager.pinTsFileResource(resource, true, PIPE_NAME); + Assert.assertTrue(pinnedModFile); + final File pipeTsFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir(resource.getTsFile(), PIPE_NAME); + final File pipeMods = new File(pipeTsFile + ModificationFile.FILE_SUFFIX); + Assert.assertEquals(1, pipeTsFileResourceManager.getFileReferenceCount(pipeMods, PIPE_NAME)); + + Assert.assertTrue(resource.remove()); + Assert.assertFalse(resource.modFileExists()); + Assert.assertTrue(pipeMods.exists()); + pipeTsFileResourceManager.unpinTsFileResource(resource, PIPE_NAME, pinnedModFile); + + Assert.assertEquals(0, pipeTsFileResourceManager.getFileReferenceCount(pipeTsFile, PIPE_NAME)); + Assert.assertEquals(0, pipeTsFileResourceManager.getFileReferenceCount(pipeMods, PIPE_NAME)); + Assert.assertEquals(0, pipeTsFileResourceManager.getFileReferenceCount(pipeMods, null)); + Assert.assertFalse(pipeMods.exists()); + } + + @Test + public void testUnpinDoesNotReleaseModsCreatedAfterPinning() throws Exception { + final TsFileResource resource = new TsFileResource(new File(TS_FILE_NAME)); + resource.removeModFile(); + final boolean pinnedModFile = + pipeTsFileResourceManager.pinTsFileResource(resource, true, PIPE_NAME); + Assert.assertFalse(pinnedModFile); + + // An event can acquire mods created after historical selection pinned the TsFile alone. + resource + .getModFile() + .write(new Deletion(new PartialPath("root.lemming.**"), 0, 0, Long.MAX_VALUE)); + resource.getModFile().close(); + final File eventMods = + pipeTsFileResourceManager.increaseFileReference( + new File(resource.getModFile().getFilePath()), false, PIPE_NAME); + pipeTsFileResourceManager.unpinTsFileResource(resource, PIPE_NAME, pinnedModFile); + + Assert.assertEquals(1, pipeTsFileResourceManager.getFileReferenceCount(eventMods, PIPE_NAME)); + Assert.assertTrue(eventMods.exists()); + pipeTsFileResourceManager.decreaseFileReference(eventMods, PIPE_NAME); + Assert.assertEquals(0, pipeTsFileResourceManager.getFileReferenceCount(eventMods, PIPE_NAME)); + Assert.assertFalse(eventMods.exists()); + } + @Test public void testConcurrentIncreaseTsFile() throws Exception { assertConcurrentIncreaseFileReference(new File(TS_FILE_NAME), true); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java index 7562e54e5811b..01c6392ef6ccc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java @@ -26,12 +26,19 @@ import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex; +import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant; import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment; import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern; import org.apache.iotdb.commons.pipe.event.ProgressReportEvent; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent; +import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; +import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager; +import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; @@ -52,6 +59,7 @@ import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -60,6 +68,66 @@ public class PipeHistoricalDataRegionTsFileSourceTest { + @Test + @SuppressWarnings("unchecked") + public void testHistoricalSourceSuppliesPinnedModsAfterOriginalFileRemoval() throws Exception { + final File tempDir = Files.createTempDirectory("pipeHistoricalSourceMods").toFile(); + final PipeHistoricalDataRegionTsFileSource source = new PipeHistoricalDataRegionTsFileSource(); + final String pipeName = "historical_source_mods"; + final long creationTime = 1L; + final String resourcePipeName = + PipeTsFileResourceManager.getPipeTsFileResourcePipeName(pipeName, creationTime); + final PipeTsFileResourceManager manager = PipeDataNodeResourceManager.tsfile(); + PipeTsFileInsertionEvent event = null; + try { + final PipeParameters parameters = + new PipeParameters( + new HashMap<>( + Collections.singletonMap( + PipeSourceConstant.SOURCE_MODS_ENABLE_KEY, Boolean.TRUE.toString()))); + source.validate(new PipeParameterValidator(parameters)); + source.customize( + parameters, + () -> + new PipeTaskSourceRuntimeEnvironment( + pipeName, creationTime, 0, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0))); + Assert.assertEquals(resourcePipeName, getPrivateField(source, "pipeNameWithCreationTime")); + + final File dataDir = new File(tempDir, "unsequence/root.db/0/0"); + Assert.assertTrue(dataDir.mkdirs()); + final TsFileResource resource = + createClosedTsFileResourceWithDevices(dataDir, "1-1-0-0.tsfile", "root.db.d0"); + resource.getModFile().write(new Deletion(new PartialPath("root.db.d0.s1"), 0, 0, 1)); + resource.getModFile().close(); + Assert.assertTrue(manager.pinTsFileResource(resource, true, resourcePipeName)); + ((Set) getPrivateField(source, "resourcesWithPinnedModFile")).add(resource); + ((Set) getPrivateField(source, "filteredTsFileResources")).add(resource); + setPrivateField(source, "pendingQueue", new ArrayDeque<>(Collections.singleton(resource))); + setPrivateField(source, "hasBeenStarted", true); + setPrivateField(source, "sloppyPattern", true); + setPrivateField(source, "sloppyTimeRange", true); + + Assert.assertTrue(resource.remove()); + event = (PipeTsFileInsertionEvent) source.supply(); + Assert.assertNotNull(event); + Assert.assertTrue(event.isWithMod()); + final File pinnedTsFile = + PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir( + resource.getTsFile(), resourcePipeName); + Assert.assertEquals(pinnedTsFile, event.getTsFile()); + Assert.assertEquals( + new File(pinnedTsFile + ModificationFile.FILE_SUFFIX), event.getModFile()); + Assert.assertEquals(1, manager.getFileReferenceCount(event.getModFile(), resourcePipeName)); + Assert.assertTrue(event.getModFile().exists()); + } finally { + source.close(); + if (event != null) { + event.clearReferenceCount(getClass().getName()); + } + FileUtils.deleteFileOrDirectory(tempDir); + } + } + @Test public void testGlobalTimeRangeRespectsHistoryEnable() throws Exception { final Map attributes = new HashMap<>();