You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2025/03/06 10:43:35 UTC
(iotdb) branch master updated: IoTConsensusV2: persistent isGeneratedByPipeConsensus (#14938)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 93f0795233d IoTConsensusV2: persistent isGeneratedByPipeConsensus (#14938)
93f0795233d is described below
commit 93f0795233da49d1be8be77641071c8357e46ff1
Author: Peng Junzhi <78...@users.noreply.github.com>
AuthorDate: Thu Mar 6 18:43:30 2025 +0800
IoTConsensusV2: persistent isGeneratedByPipeConsensus (#14938)
* feat
* spotless
* support isGeneratedByPipe
* fix
* fix review
* support compaction
* fix review
* merge
* fix
* Fix ci
* convert & to &&
---
.../common/tsfile/PipeTsFileInsertionEvent.java | 6 +-
.../event/realtime/PipeRealtimeEventFactory.java | 5 +-
...oricalDataRegionTsFileAndDeletionExtractor.java | 1 -
.../listener/PipeInsertionDataNodeListener.java | 8 +-
.../plan/analyze/load/LoadTsFileAnalyzer.java | 4 +
.../plan/scheduler/load/LoadTsFileScheduler.java | 87 +++++++++++-----------
.../execute/task/CrossSpaceCompactionTask.java | 2 +-
.../execute/task/InnerSpaceCompactionTask.java | 2 +-
.../task/RepairUnsortedFileCompactionTask.java | 2 +-
.../compaction/execute/utils/CompactionUtils.java | 12 ++-
.../dataregion/tsfile/TsFileResource.java | 19 +++++
.../pipe/event/TsFileInsertionEventParserTest.java | 1 -
12 files changed, 87 insertions(+), 62 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 cf5dc782542..7321f70f933 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
@@ -95,7 +95,6 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
final String databaseNameFromDataRegion,
final TsFileResource resource,
final boolean isLoaded,
- final boolean isGeneratedByPipe,
final boolean isGeneratedByHistoricalExtractor) {
// The modFile must be copied before the event is assigned to the listening pipes
this(
@@ -104,7 +103,6 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
resource,
true,
isLoaded,
- isGeneratedByPipe,
isGeneratedByHistoricalExtractor,
null,
0,
@@ -121,7 +119,6 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
final TsFileResource resource,
final boolean isWithMod,
final boolean isLoaded,
- final boolean isGeneratedByPipe,
final boolean isGeneratedByHistoricalExtractor,
final String pipeName,
final long creationTime,
@@ -151,7 +148,7 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
resource.getSharedModFile() != null ? resource.getSharedModFile().getFile() : null;
this.isLoaded = isLoaded;
- this.isGeneratedByPipe = isGeneratedByPipe;
+ this.isGeneratedByPipe = resource.isGeneratedByPipe();
this.isGeneratedByPipeConsensus = resource.isGeneratedByPipeConsensus();
this.isGeneratedByHistoricalExtractor = isGeneratedByHistoricalExtractor;
@@ -397,7 +394,6 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
resource,
isWithMod,
isLoaded,
- isGeneratedByPipe,
isGeneratedByHistoricalExtractor,
pipeName,
creationTime,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
index 200ed98f0ed..e290f86b356 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
@@ -47,11 +47,10 @@ public class PipeRealtimeEventFactory {
final Boolean isTableModel,
final String databaseNameFromDataRegion,
final TsFileResource resource,
- final boolean isLoaded,
- final boolean isGeneratedByPipe) {
+ final boolean isLoaded) {
PipeTsFileInsertionEvent tsFileInsertionEvent =
new PipeTsFileInsertionEvent(
- isTableModel, databaseNameFromDataRegion, resource, isLoaded, isGeneratedByPipe, false);
+ isTableModel, databaseNameFromDataRegion, resource, isLoaded, false);
// if using IoTV2, assign a replicateIndex for this event
if (DataRegionConsensusImpl.getInstance() instanceof PipeConsensus
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
index d8eb3ac6b89..57f701f4c48 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java
@@ -789,7 +789,6 @@ public class PipeHistoricalDataRegionTsFileAndDeletionExtractor
resource,
shouldTransferModFile,
false,
- false,
true,
pipeName,
creationTime,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
index 13498d8ca5f..84653a65582 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
@@ -102,6 +102,7 @@ public class PipeInsertionDataNodeListener {
final TsFileResource tsFileResource,
final boolean isLoaded,
final boolean isGeneratedByPipe) {
+ tsFileResource.setGeneratedByPipe(isGeneratedByPipe);
// We don't judge whether listenToTsFileExtractorCount.get() == 0 here on purpose
// because extractors may use tsfile events when some exceptions occur in the
// insert nodes listening process.
@@ -115,12 +116,7 @@ public class PipeInsertionDataNodeListener {
assigner.publishToAssign(
PipeRealtimeEventFactory.createRealtimeEvent(
- dataRegionId,
- assigner.isTableModel(),
- databaseName,
- tsFileResource,
- isLoaded,
- isGeneratedByPipe));
+ dataRegionId, assigner.isTableModel(), databaseName, tsFileResource, isLoaded));
}
public void listenToInsertNode(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
index 32709f13ca8..832d4dbe369 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -458,6 +458,10 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
tsFileResource.updatePlanIndexes(reader.getMaxPlanIndex());
} else {
tsFileResource.deserialize();
+ // Reset tsfileResource's isGeneratedByPipe mark to prevent deserializing the wrong mark.
+ // If this tsfile is loaded by a pipe receiver, the correct mark will be added in
+ // `listenToTsFile`
+ tsFileResource.setGeneratedByPipe(isGeneratedByPipe);
}
return tsFileResource;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
index 6a4c3839b85..2faec9a1833 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
@@ -58,7 +58,6 @@ import org.apache.iotdb.db.queryengine.plan.scheduler.FragInstanceDispatchResult
import org.apache.iotdb.db.queryengine.plan.scheduler.IScheduler;
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.storageengine.StorageEngine;
-import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex;
@@ -97,6 +96,7 @@ import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CancellationException;
@@ -521,48 +521,51 @@ public class LoadTsFileScheduler implements IScheduler {
}
// add metrics
- DataRegion dataRegion =
- StorageEngine.getInstance()
- .getDataRegion(
- (DataRegionId)
- ConsensusGroupId.Factory.createFromTConsensusGroupId(
- node.getLocalRegionReplicaSet().getRegionId()));
-
- dataRegion
- .getNonSystemDatabaseName()
+ Optional.ofNullable(
+ StorageEngine.getInstance()
+ .getDataRegion(
+ (DataRegionId)
+ ConsensusGroupId.Factory.createFromTConsensusGroupId(
+ node.getLocalRegionReplicaSet().getRegionId())))
.ifPresent(
- databaseName -> {
- // Report load tsFile points to IoTDB flush metrics
- MemTableFlushTask.recordFlushPointsMetricInternal(
- node.getWritePointCount(), databaseName, dataRegion.getDataRegionId());
-
- MetricService.getInstance()
- .count(
- node.getWritePointCount(),
- Metric.QUANTITY.toString(),
- MetricLevel.CORE,
- Tag.NAME.toString(),
- Metric.POINTS_IN.toString(),
- Tag.DATABASE.toString(),
- databaseName,
- Tag.REGION.toString(),
- dataRegion.getDataRegionId(),
- Tag.TYPE.toString(),
- Metric.LOAD_POINT_COUNT.toString());
- MetricService.getInstance()
- .count(
- node.getWritePointCount(),
- Metric.LEADER_QUANTITY.toString(),
- MetricLevel.CORE,
- Tag.NAME.toString(),
- Metric.POINTS_IN.toString(),
- Tag.DATABASE.toString(),
- databaseName,
- Tag.REGION.toString(),
- dataRegion.getDataRegionId(),
- Tag.TYPE.toString(),
- Metric.LOAD_POINT_COUNT.toString());
- });
+ dataRegion ->
+ dataRegion
+ .getNonSystemDatabaseName()
+ .ifPresent(
+ databaseName -> {
+ // Report load tsFile points to IoTDB flush metrics
+ MemTableFlushTask.recordFlushPointsMetricInternal(
+ node.getWritePointCount(),
+ databaseName,
+ dataRegion.getDataRegionId());
+
+ MetricService.getInstance()
+ .count(
+ node.getWritePointCount(),
+ Metric.QUANTITY.toString(),
+ MetricLevel.CORE,
+ Tag.NAME.toString(),
+ Metric.POINTS_IN.toString(),
+ Tag.DATABASE.toString(),
+ databaseName,
+ Tag.REGION.toString(),
+ dataRegion.getDataRegionId(),
+ Tag.TYPE.toString(),
+ Metric.LOAD_POINT_COUNT.toString());
+ MetricService.getInstance()
+ .count(
+ node.getWritePointCount(),
+ Metric.LEADER_QUANTITY.toString(),
+ MetricLevel.CORE,
+ Tag.NAME.toString(),
+ Metric.POINTS_IN.toString(),
+ Tag.DATABASE.toString(),
+ databaseName,
+ Tag.REGION.toString(),
+ dataRegion.getDataRegionId(),
+ Tag.TYPE.toString(),
+ Metric.LOAD_POINT_COUNT.toString());
+ }));
return true;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
index 04e3ccb4b27..41d859924f0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
@@ -193,7 +193,7 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
performer.setSummary(summary);
performer.perform();
- CompactionUtils.updateProgressIndex(
+ CompactionUtils.updateProgressIndexAndMark(
targetTsfileResourceList, selectedSequenceFiles, selectedUnsequenceFiles);
CompactionUtils.moveTargetFile(
targetTsfileResourceList,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
index e75f4d9ba93..21c62232e8e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
@@ -439,7 +439,7 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
}
protected void prepareTargetFiles() throws IOException {
- CompactionUtils.updateProgressIndex(
+ CompactionUtils.updateProgressIndexAndMark(
filesView.targetFilesInPerformer,
filesView.sequence ? filesView.sourceFilesInCompactionPerformer : Collections.emptyList(),
filesView.sequence ? Collections.emptyList() : filesView.sourceFilesInCompactionPerformer);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/RepairUnsortedFileCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/RepairUnsortedFileCompactionTask.java
index f04103fdb3e..8619c4c04f7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/RepairUnsortedFileCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/RepairUnsortedFileCompactionTask.java
@@ -151,7 +151,7 @@ public class RepairUnsortedFileCompactionTask extends InnerSpaceCompactionTask {
@Override
protected void prepareTargetFiles() throws IOException {
- CompactionUtils.updateProgressIndex(
+ CompactionUtils.updateProgressIndexAndMark(
filesView.targetFilesInPerformer,
filesView.sourceFilesInCompactionPerformer,
Collections.emptyList());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
index 6e1eafd1cf0..44905afc461 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
@@ -324,16 +324,26 @@ public class CompactionUtils {
}
}
- public static void updateProgressIndex(
+ public static void updateProgressIndexAndMark(
List<TsFileResource> targetResources,
List<TsFileResource> seqResources,
List<TsFileResource> unseqResources) {
for (TsFileResource targetResource : targetResources) {
for (TsFileResource unseqResource : unseqResources) {
targetResource.updateProgressIndex(unseqResource.getMaxProgressIndexAfterClose());
+ targetResource.setGeneratedByPipe(
+ unseqResource.isGeneratedByPipe() && targetResource.isGeneratedByPipe());
+ targetResource.setGeneratedByPipeConsensus(
+ unseqResource.isGeneratedByPipeConsensus()
+ && targetResource.isGeneratedByPipeConsensus());
}
for (TsFileResource seqResource : seqResources) {
targetResource.updateProgressIndex(seqResource.getMaxProgressIndexAfterClose());
+ targetResource.setGeneratedByPipe(
+ seqResource.isGeneratedByPipe() && targetResource.isGeneratedByPipe());
+ targetResource.setGeneratedByPipeConsensus(
+ seqResource.isGeneratedByPipeConsensus()
+ && targetResource.isGeneratedByPipeConsensus());
}
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index 5b373ba268d..204fe1f32d4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -199,6 +199,9 @@ public class TsFileResource implements PersistentResource {
/** used to prevent circular replication in PipeConsensus */
private boolean isGeneratedByPipeConsensus = false;
+ /** used to prevent circular replication in Pipe */
+ private boolean isGeneratedByPipe = false;
+
private InsertionCompactionCandidateStatus insertionCompactionCandidateStatus =
InsertionCompactionCandidateStatus.NOT_CHECKED;
@@ -302,6 +305,9 @@ public class TsFileResource implements PersistentResource {
} else {
TsFileResourceBlockType.EMPTY_BLOCK.serialize(outputStream);
}
+
+ ReadWriteIOUtils.write(isGeneratedByPipeConsensus, outputStream);
+ ReadWriteIOUtils.write(isGeneratedByPipe, outputStream);
}
/** deserialize from disk */
@@ -333,6 +339,11 @@ public class TsFileResource implements PersistentResource {
maxProgressIndex = ProgressIndexType.deserializeFrom(inputStream);
}
}
+
+ if (inputStream.available() > 0) {
+ isGeneratedByPipeConsensus = ReadWriteIOUtils.readBoolean(inputStream);
+ isGeneratedByPipe = ReadWriteIOUtils.readBoolean(inputStream);
+ }
}
}
@@ -741,6 +752,14 @@ public class TsFileResource implements PersistentResource {
isGeneratedByPipeConsensus = generatedByPipeConsensus;
}
+ public boolean isGeneratedByPipe() {
+ return isGeneratedByPipe;
+ }
+
+ public void setGeneratedByPipe(boolean generatedByPipe) {
+ isGeneratedByPipe = generatedByPipe;
+ }
+
public void writeLock() {
if (originTsFileResource == null) {
tsFileLock.writeLock();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
index da87b705213..f0710f6a86d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/TsFileInsertionEventParserTest.java
@@ -566,7 +566,6 @@ public class TsFileInsertionEventParserTest {
true,
false,
false,
- false,
null,
0,
null,