IoTV2: Fix the issue of resend lots of replicated files when DN restarts.

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java

@@ -21,6 +21,9 @@
 
 import org.apache.iotdb.commons.consensus.DataRegionId;
 import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.consensus.index.ProgressIndexType;
+import org.apache.iotdb.commons.consensus.index.impl.HybridProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.StateProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex;
 import org.apache.iotdb.commons.exception.IllegalPathException;
@@ -34,6 +37,7 @@
 import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.consensus.pipe.PipeConsensus;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
 import org.apache.iotdb.db.pipe.consensus.ReplicateProgressDataNodeManager;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
@@ -304,7 +308,12 @@ public void customize(
  pipeName = environment.getPipeName();
  creationTime = environment.getCreationTime();
  pipeTaskMeta = environment.getPipeTaskMeta();
- startIndex = environment.getPipeTaskMeta().getProgressIndex();
+ if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
+ startIndex =
+ tryToExtractLocalProgressIndexForIoTV2(environment.getPipeTaskMeta().getProgressIndex());
+ } else {
+ startIndex = environment.getPipeTaskMeta().getProgressIndex();
+ }
 
  dataRegionId = environment.getRegionId();
  synchronized (DATA_REGION_ID_TO_PIPE_FLUSHED_TIME_MAP) {
@@ -426,6 +435,61 @@ private void flushDataRegionAllTsFiles() {
  }
  }
 
+ /**
+ * IoTV2 will only resend event that contains un-replicated local write data. So we only extract
+ * ProgressIndex containing local writes for comparison to prevent misjudgment on whether
+ * high-level tsFiles with mixed progressIndexes need to be retransmitted
+ *
+ * @return recoverProgressIndex dedicated in local DataNodeId or origin for fallback.
+ */
+ private ProgressIndex tryToExtractLocalProgressIndexForIoTV2(ProgressIndex origin) {
+ // There are only 2 cases:
+ // 1. origin is RecoverProgressIndex
+ if (origin instanceof RecoverProgressIndex) {
+ RecoverProgressIndex toBeTransformed = (RecoverProgressIndex) origin;
+ return extractRecoverProgressIndex(toBeTransformed);
+ }
+ // 2. origin is HybridProgressIndex
+ else if (origin instanceof HybridProgressIndex) {
+ HybridProgressIndex toBeTransformed = (HybridProgressIndex) origin;
+ // if hybridProgressIndex contains recoverProgressIndex, which is what we expected.
+ if (toBeTransformed
+ .getType2Index()
+ .containsKey(ProgressIndexType.RECOVER_PROGRESS_INDEX.getType())) {
+ // 2.1. transform recoverProgressIndex
+ RecoverProgressIndex specificToBeTransformed =
+ (RecoverProgressIndex)
+ toBeTransformed
+ .getType2Index()
+ .get(ProgressIndexType.RECOVER_PROGRESS_INDEX.getType());
+ return extractRecoverProgressIndex(specificToBeTransformed);
+ }
+ // if hybridProgressIndex doesn't contain recoverProgressIndex, which is not what we expected,
+ // fallback.
+ return origin;
+ } else {
+ // fallback
+ LOGGER.warn(
+ "Pipe {}@{}: unexpected ProgressIndex type {}, fallback to origin {}.",
+ pipeName,
+ dataRegionId,
+ origin.getType(),
+ origin);
+ return origin;
+ }
+ }
+
+ private ProgressIndex extractRecoverProgressIndex(RecoverProgressIndex toBeTransformed) {
+ return new RecoverProgressIndex(
+ toBeTransformed.getDataNodeId2LocalIndex().entrySet().stream()
+ .filter(
+ entry ->
+ entry
+ .getKey()
+ .equals(IoTDBDescriptor.getInstance().getConfig().getDataNodeId()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+ }
+
  @Override
  public synchronized void start() {
  if (!shouldExtractInsertion) {
@@ -617,8 +681,19 @@ private boolean mayTsFileContainUnprocessedData(final TsFileResource resource) {
  if (startIndex instanceof StateProgressIndex) {
  startIndex = ((StateProgressIndex) startIndex).getInnerProgressIndex();
  }
- return !startIndex.isAfter(resource.getMaxProgressIndexAfterClose())
- && !startIndex.equals(resource.getMaxProgressIndexAfterClose());
+
+ if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
+ // For consensus pipe, we only focus on the progressIndex that is generated from local write
+ // instead of replication or something else.
+ ProgressIndex dedicatedProgressIndex =
+ tryToExtractLocalProgressIndexForIoTV2(resource.getMaxProgressIndexAfterClose());
+ return greaterThanStartIndex(dedicatedProgressIndex);
+ }
+ return greaterThanStartIndex(resource.getMaxProgressIndexAfterClose());
+ }
+
+ private boolean greaterThanStartIndex(ProgressIndex progressIndex) {
+ return !startIndex.isAfter(progressIndex) && !startIndex.equals(progressIndex);
  }
 
  private boolean mayTsFileResourceOverlappedWithPattern(final TsFileResource resource) {
@@ -712,12 +787,26 @@ private void extractDeletions(
  // For deletions that are filtered and will not be sent, we should manually decrease its
  // reference count. Because the initial value of referenceCount is `ReplicaNum - 1`
  allDeletionResources.stream()
- .filter(resource -> startIndex.isAfter(resource.getProgressIndex()))
+ .filter(
+ resource -> {
+ ProgressIndex toBeCompared = resource.getProgressIndex();
+ if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
+ toBeCompared = tryToExtractLocalProgressIndexForIoTV2(toBeCompared);
+ }
+ return !greaterThanStartIndex(toBeCompared);
+ })
  .forEach(DeletionResource::decreaseReference);
  // Get deletions that should be sent.
  allDeletionResources =
  allDeletionResources.stream()
- .filter(resource -> !startIndex.isAfter(resource.getProgressIndex()))
+ .filter(
+ resource -> {
+ ProgressIndex toBeCompared = resource.getProgressIndex();
+ if (pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)) {
+ toBeCompared = tryToExtractLocalProgressIndexForIoTV2(toBeCompared);
+ }
+ return greaterThanStartIndex(toBeCompared);
+ })
  .collect(Collectors.toList());
  resourceList.addAll(allDeletionResources);
  LOGGER.info(