You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/08/06 13:06:24 UTC
[iotdb] branch iotdb-6101-1.2 created (now 6e6eef73eb0)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a change to branch iotdb-6101-1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 6e6eef73eb0 [IOTDB-6101] Pipe: Support tsfile cascade transport (#10795)
This branch includes the following new commits:
new 6e6eef73eb0 [IOTDB-6101] Pipe: Support tsfile cascade transport (#10795)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: [IOTDB-6101] Pipe: Support tsfile cascade transport (#10795)
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch iotdb-6101-1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6e6eef73eb0ce2e1d7c60b0f0b53e7488d9b3489
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Sun Aug 6 21:02:30 2023 +0800
[IOTDB-6101] Pipe: Support tsfile cascade transport (#10795)
Support tsfile cascade transport. For example, there are 3 iotdb clusters A, B and C. Now we can use pipe to transport tsfile from A to C (via B, A -> B -> C).
(cherry picked from commit b3a4bdf813f7f307a4a8b06aab73432da773dcda)
---
.../listener/PipeInsertionDataNodeListener.java | 9 +++------
.../planner/plan/node/load/LoadSingleTsFileNode.java | 11 ++++++++---
.../iotdb/db/storageengine/dataregion/DataRegion.java | 5 +++++
.../dataregion/memtable/TsFileProcessor.java | 2 ++
.../org/apache/iotdb/db/utils/FileLoaderUtils.java | 2 ++
.../dataregion/TsFileResourceProgressIndexTest.java | 18 ++++++++++++++++++
6 files changed, 38 insertions(+), 9 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
index 8afa1831566..b7a7a0c3770 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/realtime/listener/PipeInsertionDataNodeListener.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.pipe.extractor.realtime.listener;
-import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
import org.apache.iotdb.db.pipe.extractor.realtime.PipeRealtimeDataRegionExtractor;
import org.apache.iotdb.db.pipe.extractor.realtime.assigner.PipeDataRegionAssigner;
@@ -93,11 +92,9 @@ public class PipeInsertionDataNodeListener {
//////////////////////////// listen to events ////////////////////////////
public void listenToTsFile(String dataRegionId, TsFileResource tsFileResource) {
- // wo don't judge whether listenToTsFileExtractorCount.get() == 0 here, because
- // when using SimpleProgressIndex, the tsfile event needs to be assigned to the
- // extractor even if listenToTsFileExtractorCount.get() == 0 to record the progress
-
- PipeAgent.runtime().assignSimpleProgressIndexIfNeeded(tsFileResource);
+ // 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.
final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
index d536a357412..4f3506bcb75 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -99,9 +100,13 @@ public class LoadSingleTsFileNode extends WritePlanNode {
needDecodeTsFile = !isDispatchedToLocal(new HashSet<>(partitionFetcher.apply(slotList)));
}
- if (!needDecodeTsFile && !resource.resourceFileExists()) {
- resource.serialize();
- }
+ PipeAgent.runtime().assignRecoverProgressIndexForTsFileRecovery(resource);
+
+ // we serialize the resource file even if the tsfile does not need to be decoded
+ // or the resource file is already existed because we need to serialize the
+ // progress index of the tsfile
+ resource.serialize();
+
return needDecodeTsFile;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index db5d0168d20..0011db21534 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -47,6 +47,7 @@ import org.apache.iotdb.db.exception.WriteProcessRejectException;
import org.apache.iotdb.db.exception.query.OutOfTTLException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.exception.quota.ExceedQuotaException;
+import org.apache.iotdb.db.pipe.extractor.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
@@ -2221,6 +2222,9 @@ public class DataRegion implements IDataRegionForQuery {
}
loadTsFileToUnSequence(
tsfileToBeInserted, newTsFileResource, newFilePartitionId, deleteOriginFile);
+
+ PipeInsertionDataNodeListener.getInstance().listenToTsFile(dataRegionId, newTsFileResource);
+
FileMetrics.getInstance()
.addFile(
newTsFileResource.getTsFile().length(),
@@ -2429,6 +2433,7 @@ public class DataRegion implements IDataRegionForQuery {
} else {
Files.copy(resourceFileToLoad.toPath(), targetResourceFile.toPath());
}
+
} catch (IOException e) {
logger.error(
"File renaming failed when loading .resource file. Origin: {}, Target: {}",
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index c82cc639434..52352408b19 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.WriteProcessRejectException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.pipe.extractor.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet;
@@ -862,6 +863,7 @@ public class TsFileProcessor {
IMemTable tmpMemTable = workMemTable == null ? new NotifyFlushMemTable() : workMemTable;
try {
+ PipeAgent.runtime().assignSimpleProgressIndexIfNeeded(tsFileResource);
PipeInsertionDataNodeListener.getInstance()
.listenToTsFile(dataRegionInfo.getDataRegion().getDataRegionId(), tsFileResource);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
index 28bb8f82930..2f243b2b508 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.utils;
import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet;
import org.apache.iotdb.db.storageengine.buffer.TimeSeriesMetadataCache;
@@ -106,6 +107,7 @@ public class FileLoaderUtils {
}
}
resource.setStatus(TsFileResourceStatus.NORMAL);
+ PipeAgent.runtime().assignRecoverProgressIndexForTsFileRecovery(resource);
return resource;
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java
index c5339c1b738..62b9caa2f35 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/TsFileResourceProgressIndexTest.java
@@ -21,6 +21,9 @@ package org.apache.iotdb.db.storageengine.dataregion;
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.SimpleProgressIndex;
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.generator.TsFileNameGenerator;
@@ -87,6 +90,21 @@ public class TsFileResourceProgressIndexTest {
@Test
public void testProgressIndexRecorder() {
+ HybridProgressIndex hybridProgressIndex = new HybridProgressIndex();
+ hybridProgressIndex.updateToMinimumIsAfterProgressIndex(new SimpleProgressIndex(3, 4));
+ hybridProgressIndex.updateToMinimumIsAfterProgressIndex(new SimpleProgressIndex(6, 6));
+ hybridProgressIndex.updateToMinimumIsAfterProgressIndex(
+ new RecoverProgressIndex(1, new SimpleProgressIndex(1, 2)));
+ hybridProgressIndex.updateToMinimumIsAfterProgressIndex(
+ new RecoverProgressIndex(1, new SimpleProgressIndex(1, 3)));
+ hybridProgressIndex.updateToMinimumIsAfterProgressIndex(
+ new RecoverProgressIndex(2, new SimpleProgressIndex(4, 3)));
+ hybridProgressIndex.updateToMinimumIsAfterProgressIndex(
+ new RecoverProgressIndex(3, new SimpleProgressIndex(5, 5)));
+ Assert.assertTrue(hybridProgressIndex.isAfter(new SimpleProgressIndex(6, 5)));
+ Assert.assertTrue(
+ hybridProgressIndex.isAfter(new RecoverProgressIndex(3, new SimpleProgressIndex(5, 4))));
+
Assert.assertTrue(
new MockProgressIndex(0).isAfter(tsFileResource.getMaxProgressIndexAfterClose()));