Skip to content

[To dev/1.3] [Pipe] Preserve historical TsFile mods after compaction (#18904) #18923

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: dev/1.3
Choose a base branch
from
Open

[To dev/1.3] [Pipe] Preserve historical TsFile mods after compaction (#18904) #18923

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 17 additions & 1 deletion .../src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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",
Expand Down
38 changes: 22 additions & 16 deletions ...ode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,8 @@ public Map<String, TSDataType> 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);
Expand All @@ -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).
Expand All @@ -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;
Expand All @@ -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);
}
}
}

Expand Down
22 changes: 18 additions & 4 deletions ...ache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -109,6 +110,7 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR

private String pipeName;
private long creationTime;
private String pipeNameWithCreationTime;

private PipeTaskMeta pipeTaskMeta;
private ProgressIndex startIndex;
Expand Down Expand Up @@ -141,6 +143,7 @@ public class PipeHistoricalDataRegionTsFileSource implements PipeHistoricalDataR

private Queue<TsFileResource> pendingQueue;
private final Set<TsFileResource> filteredTsFileResources = new HashSet<>();
private final Set<TsFileResource> resourcesWithPinnedModFile = new HashSet<>();
private final Set<TsFileResource> historicalProgressReportResources = new HashSet<>();
private ProgressIndex maxHistoricalProgressIndex = MinimumProgressIndex.INSTANCE;
private ProgressIndex maxSuppliedHistoricalProgressReportIndex = MinimumProgressIndex.INSTANCE;
Expand Down Expand Up @@ -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();

Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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: {}",
Expand Down Expand Up @@ -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: {}",
Expand All @@ -931,6 +944,7 @@ public synchronized void close() {
pendingQueue.clear();
pendingQueue = null;
}
resourcesWithPinnedModFile.clear();
historicalProgressReportResources.clear();
pendingHistoricalProgressIndexToReport = null;
}
Expand Down
97 changes: 97 additions & 0 deletions ...e/datanode/src/test/java/org/apache/iotdb/db/pipe/event/PipeTsFileInsertionEventTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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();
Expand Down
48 changes: 48 additions & 0 deletions ...tanode/src/test/java/org/apache/iotdb/db/pipe/resource/PipeTsFileResourceManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
Loading
Loading