You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ju...@apache.org on 2026/02/13 10:56:25 UTC
(iotdb) branch pipe-flush updated (7cbf7d93540 -> 086f9793ff2)
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a change to branch pipe-flush
in repository https://gitbox.apache.org/repos/asf/iotdb.git
discard 7cbf7d93540 gsa
discard dea7feb59eb may-fix
discard 7b637991817 hybrid
discard 6ee24e24dff refactor
discard 590831ce426 partial
new 7bbdc95f844 Flush manager
new 8eac942c998 IT-fix
new 086f9793ff2 clean
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (7cbf7d93540)
\
N -- N -- N refs/heads/pipe-flush (086f9793ff2)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 3 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.
Summary of changes:
.../it/env/cluster/config/MppCommonConfig.java | 7 +++
.../env/cluster/config/MppSharedCommonConfig.java | 7 +++
.../it/env/remote/config/RemoteCommonConfig.java | 5 ++
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 +
.../treemodel/auto/basic/IoTDBPipeSourceIT.java | 53 +++++++++++++++++++++-
.../agent/task/subtask/sink/PipeSinkSubtask.java | 3 +-
.../pipe/consensus/deletion/DeletionResource.java | 9 ++--
.../deletion/DeletionResourceManager.java | 8 ++--
.../deletion/persist/PageCacheDeletionBuffer.java | 4 +-
.../consensus/deletion/recover/DeletionReader.java | 4 +-
.../event/common/heartbeat/PipeHeartbeatEvent.java | 6 +--
.../event/realtime/PipeRealtimeEventFactory.java | 2 +-
.../db/pipe/metric/source/PipeAssignerMetrics.java | 32 ++++++-------
...istoricalDataRegionTsFileAndDeletionSource.java | 4 +-
.../realtime/PipeRealtimeDataRegionSource.java | 4 +-
.../realtime/assigner/PipeDataRegionAssigner.java | 8 ++--
.../listener/PipeInsertionDataNodeListener.java | 26 +++++------
.../db/storageengine/dataregion/DataRegion.java | 19 +++++---
.../dataregion/memtable/TsFileProcessor.java | 8 ++--
.../db/pipe/consensus/DeletionRecoverTest.java | 6 +--
.../db/pipe/consensus/DeletionResourceTest.java | 22 ++++-----
.../db/pipe/source/PipeRealtimeExtractTest.java | 32 ++++---------
22 files changed, 166 insertions(+), 105 deletions(-)
(iotdb) 03/03: clean
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch pipe-flush
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 086f9793ff22f898fd206c2b0416b908db25b845
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Feb 13 17:40:26 2026 +0800
clean
---
.../listener/PipeInsertionDataNodeListener.java | 18 ++++++------
.../db/storageengine/dataregion/DataRegion.java | 8 ++++--
.../dataregion/memtable/TsFileProcessor.java | 8 +++---
.../db/pipe/source/PipeRealtimeExtractTest.java | 32 +++++++---------------
4 files changed, 29 insertions(+), 37 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
index 8fb95d78fc4..d255d80166a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
@@ -51,8 +51,8 @@ public class PipeInsertionDataNodeListener {
private final ConcurrentMap<Integer, PipeDataRegionAssigner> dataRegionId2Assigner =
new ConcurrentHashMap<>();
- private final AtomicInteger listenToTsFileExtractorCount = new AtomicInteger(0);
- private final AtomicInteger listenToInsertNodeExtractorCount = new AtomicInteger(0);
+ private final AtomicInteger listenToTsFileSourceCount = new AtomicInteger(0);
+ private final AtomicInteger listenToInsertNodeSourceCount = new AtomicInteger(0);
//////////////////////////// start & stop ////////////////////////////
@@ -63,10 +63,10 @@ public class PipeInsertionDataNodeListener {
.startAssignTo(extractor);
if (extractor.isNeedListenToTsFile()) {
- listenToTsFileExtractorCount.incrementAndGet();
+ listenToTsFileSourceCount.incrementAndGet();
}
if (extractor.isNeedListenToInsertNode()) {
- listenToInsertNodeExtractorCount.incrementAndGet();
+ listenToInsertNodeSourceCount.incrementAndGet();
}
}
@@ -80,10 +80,10 @@ public class PipeInsertionDataNodeListener {
assigner.stopAssignTo(extractor);
if (extractor.isNeedListenToTsFile()) {
- listenToTsFileExtractorCount.decrementAndGet();
+ listenToTsFileSourceCount.decrementAndGet();
}
if (extractor.isNeedListenToInsertNode()) {
- listenToInsertNodeExtractorCount.decrementAndGet();
+ listenToInsertNodeSourceCount.decrementAndGet();
}
if (assigner.notMoreExtractorNeededToBeAssigned()) {
@@ -97,7 +97,7 @@ public class PipeInsertionDataNodeListener {
//////////////////////////// listen to events ////////////////////////////
public void listenToTsFile(
- final String dataRegionId,
+ final int dataRegionId,
final String databaseName,
final TsFileResource tsFileResource,
final boolean isLoaded) {
@@ -118,11 +118,11 @@ public class PipeInsertionDataNodeListener {
}
public void listenToInsertNode(
- final String dataRegionId,
+ final int dataRegionId,
final String databaseName,
final InsertNode insertNode,
final TsFileResource tsFileResource) {
- if (listenToInsertNodeExtractorCount.get() == 0) {
+ if (listenToInsertNodeSourceCount.get() == 0) {
return;
}
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 8a76f891679..7a0d5b86833 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
@@ -4157,7 +4157,7 @@ public class DataRegion implements IDataRegionForQuery {
// Listen before the tsFile is added into tsFile manager to avoid it being compacted
PipeInsertionDataNodeListener.getInstance()
- .listenToTsFile(dataRegionIdString, databaseName, tsFileResource, true);
+ .listenToTsFile(dataRegionId.getId(), databaseName, tsFileResource, true);
tsFileManager.add(tsFileResource, false);
@@ -4336,12 +4336,16 @@ public class DataRegion implements IDataRegionForQuery {
return dataRegionIdString;
}
+ public int getDataRegionId() {
+ return dataRegionId.getId();
+ }
+
/**
* Get the storageGroupPath with dataRegionId.
*
* @return data region path, like root.sg1/0
*/
- public String getStorageGroupPath() {
+ public String getDatabasePath() {
return databaseName + File.separator + dataRegionIdString;
}
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 acdac61180b..85e0bbba055 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
@@ -340,7 +340,7 @@ public class TsFileProcessor {
}
PipeInsertionDataNodeListener.getInstance()
.listenToInsertNode(
- dataRegionInfo.getDataRegion().getDataRegionIdString(),
+ dataRegionInfo.getDataRegion().getDataRegionId(),
dataRegionInfo.getDataRegion().getDatabaseName(),
insertRowNode,
tsFileResource);
@@ -437,7 +437,7 @@ public class TsFileProcessor {
}
PipeInsertionDataNodeListener.getInstance()
.listenToInsertNode(
- dataRegionInfo.getDataRegion().getDataRegionIdString(),
+ dataRegionInfo.getDataRegion().getDataRegionId(),
dataRegionInfo.getDataRegion().getDatabaseName(),
insertRowsNode,
tsFileResource);
@@ -610,7 +610,7 @@ public class TsFileProcessor {
}
PipeInsertionDataNodeListener.getInstance()
.listenToInsertNode(
- dataRegionInfo.getDataRegion().getDataRegionIdString(),
+ dataRegionInfo.getDataRegion().getDataRegionId(),
dataRegionInfo.getDataRegion().getDatabaseName(),
insertTabletNode,
tsFileResource);
@@ -1746,7 +1746,7 @@ public class TsFileProcessor {
// before resource serialization to avoid missing hardlink after restart
PipeInsertionDataNodeListener.getInstance()
.listenToTsFile(
- dataRegionInfo.getDataRegion().getDataRegionIdString(),
+ dataRegionInfo.getDataRegion().getDataRegionId(),
dataRegionInfo.getDataRegion().getDatabaseName(),
tsFileResource,
false);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
index 59aec4b1f7b..9e07e91e983 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/PipeRealtimeExtractTest.java
@@ -70,8 +70,8 @@ public class PipeRealtimeExtractTest {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeRealtimeExtractTest.class);
- private final String dataRegion1 = "1";
- private final String dataRegion2 = "2";
+ private final int dataRegion1 = 1;
+ private final int dataRegion2 = 2;
private final String pattern1 = "root.sg.d";
private final String pattern2 = "root.sg.d.a";
private final String[] device = new String[] {"root", "sg", "d"};
@@ -151,31 +151,19 @@ public class PipeRealtimeExtractTest {
final PipeTaskRuntimeConfiguration configuration0 =
new PipeTaskRuntimeConfiguration(
new PipeTaskSourceRuntimeEnvironment(
- "1",
- 1,
- Integer.parseInt(dataRegion1),
- new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
+ "1", 1, dataRegion1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
final PipeTaskRuntimeConfiguration configuration1 =
new PipeTaskRuntimeConfiguration(
new PipeTaskSourceRuntimeEnvironment(
- "1",
- 1,
- Integer.parseInt(dataRegion1),
- new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
+ "1", 1, dataRegion1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
final PipeTaskRuntimeConfiguration configuration2 =
new PipeTaskRuntimeConfiguration(
new PipeTaskSourceRuntimeEnvironment(
- "1",
- 1,
- Integer.parseInt(dataRegion2),
- new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
+ "1", 1, dataRegion2, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
final PipeTaskRuntimeConfiguration configuration3 =
new PipeTaskRuntimeConfiguration(
new PipeTaskSourceRuntimeEnvironment(
- "1",
- 1,
- Integer.parseInt(dataRegion2),
- new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
+ "1", 1, dataRegion2, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1)));
// Some parameters of extractor are validated and initialized during the validation process.
extractor0.validate(new PipeParameterValidator(parameters0));
@@ -274,7 +262,7 @@ public class PipeRealtimeExtractTest {
}
private Future<?> write2DataRegion(
- final int writeNum, final String dataRegionId, final int startNum) {
+ final int writeNum, final int dataRegionId, final int startNum) {
final File dataRegionDir =
new File(tsFileDir.getPath() + File.separator + dataRegionId + File.separator + "0");
final boolean ignored = dataRegionDir.mkdirs();
@@ -305,7 +293,7 @@ public class PipeRealtimeExtractTest {
PipeInsertionDataNodeListener.getInstance()
.listenToInsertNode(
dataRegionId,
- dataRegionId,
+ Integer.toString(dataRegionId),
new InsertRowNode(
new PlanNodeId(String.valueOf(i)),
new PartialPath(device),
@@ -319,7 +307,7 @@ public class PipeRealtimeExtractTest {
PipeInsertionDataNodeListener.getInstance()
.listenToInsertNode(
dataRegionId,
- dataRegionId,
+ Integer.toString(dataRegionId),
new InsertRowNode(
new PlanNodeId(String.valueOf(i)),
new PartialPath(device),
@@ -331,7 +319,7 @@ public class PipeRealtimeExtractTest {
false),
resource);
PipeInsertionDataNodeListener.getInstance()
- .listenToTsFile(dataRegionId, dataRegionId, resource, false);
+ .listenToTsFile(dataRegionId, Integer.toString(dataRegionId), resource, false);
}
});
}
(iotdb) 02/03: IT-fix
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch pipe-flush
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8eac942c9984c03f2b4713bc75ecbff5fae23615
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Feb 13 17:22:54 2026 +0800
IT-fix
---
.../it/env/cluster/config/MppCommonConfig.java | 7 +++
.../env/cluster/config/MppSharedCommonConfig.java | 7 +++
.../it/env/remote/config/RemoteCommonConfig.java | 5 ++
.../org/apache/iotdb/itbase/env/CommonConfig.java | 2 +
.../treemodel/auto/basic/IoTDBPipeSourceIT.java | 53 +++++++++++++++++++++-
.../agent/task/subtask/sink/PipeSinkSubtask.java | 3 +-
.../pipe/consensus/deletion/DeletionResource.java | 9 ++--
.../deletion/DeletionResourceManager.java | 8 ++--
.../deletion/persist/PageCacheDeletionBuffer.java | 4 +-
.../consensus/deletion/recover/DeletionReader.java | 4 +-
.../event/common/heartbeat/PipeHeartbeatEvent.java | 6 +--
.../event/realtime/PipeRealtimeEventFactory.java | 2 +-
.../db/pipe/metric/source/PipeAssignerMetrics.java | 32 ++++++-------
...istoricalDataRegionTsFileAndDeletionSource.java | 4 +-
.../realtime/PipeRealtimeDataRegionSource.java | 4 +-
.../realtime/assigner/PipeDataRegionAssigner.java | 8 ++--
.../listener/PipeInsertionDataNodeListener.java | 8 ++--
.../db/storageengine/dataregion/DataRegion.java | 11 +++--
.../db/pipe/consensus/DeletionRecoverTest.java | 6 +--
.../db/pipe/consensus/DeletionResourceTest.java | 22 ++++-----
20 files changed, 137 insertions(+), 68 deletions(-)
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
index 53aa395491b..f2b788b772f 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java
@@ -465,6 +465,13 @@ public class MppCommonConfig extends MppBaseConfig implements CommonConfig {
return this;
}
+ @Override
+ public CommonConfig setPipeTsFileFlushIntervalSeconds(long pipeTsFileFlushIntervalSeconds) {
+ setProperty(
+ "pipe_tsfile_flush_interval_seconds", String.valueOf(pipeTsFileFlushIntervalSeconds));
+ return this;
+ }
+
@Override
public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) {
setProperty("pipe_enable_memory_checked", String.valueOf(isPipeEnableMemoryCheck));
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
index ed8a8943303..88a85e27f46 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
@@ -479,6 +479,13 @@ public class MppSharedCommonConfig implements CommonConfig {
return this;
}
+ @Override
+ public CommonConfig setPipeTsFileFlushIntervalSeconds(long pipeTsFileFlushIntervalSeconds) {
+ dnConfig.setPipeTsFileFlushIntervalSeconds(pipeTsFileFlushIntervalSeconds);
+ cnConfig.setPipeTsFileFlushIntervalSeconds(pipeTsFileFlushIntervalSeconds);
+ return this;
+ }
+
@Override
public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) {
dnConfig.setIsPipeEnableMemoryCheck(isPipeEnableMemoryCheck);
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
index 0c05af17f6a..31fbe9e78fb 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java
@@ -338,6 +338,11 @@ public class RemoteCommonConfig implements CommonConfig {
return this;
}
+ @Override
+ public CommonConfig setPipeTsFileFlushIntervalSeconds(long pipeTsFileFlushIntervalSeconds) {
+ return this;
+ }
+
@Override
public CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck) {
return this;
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
index 95ac239e89b..92a6af5851d 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java
@@ -150,6 +150,8 @@ public interface CommonConfig {
CommonConfig setPipeMemoryManagementEnabled(boolean pipeMemoryManagementEnabled);
+ CommonConfig setPipeTsFileFlushIntervalSeconds(long pipeTsFileFlushIntervalSeconds);
+
CommonConfig setIsPipeEnableMemoryCheck(boolean isPipeEnableMemoryCheck);
CommonConfig setPipeAirGapReceiverEnabled(boolean isPipeAirGapReceiverEnabled);
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
index a45115f5a49..3351958d2a1 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
@@ -74,6 +74,7 @@ public class IoTDBPipeSourceIT extends AbstractPipeDualTreeModelAutoIT {
.setEnableUnseqSpaceCompaction(false)
.setEnableCrossSpaceCompaction(false)
.setPipeMemoryManagementEnabled(false)
+ .setPipeTsFileFlushIntervalSeconds(10)
.setIsPipeEnableMemoryCheck(false)
.setPipeAutoSplitFullEnabled(false);
senderEnv.getConfig().getConfigNodeConfig().setLeaderDistributionPolicy("HASH");
@@ -323,7 +324,7 @@ public class IoTDBPipeSourceIT extends AbstractPipeDualTreeModelAutoIT {
}
@Test
- public void testExtractorInvalidParameter() throws Exception {
+ public void testSourceInvalidParameter() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
final String receiverIp = receiverDataNode.getIp();
@@ -608,6 +609,56 @@ public class IoTDBPipeSourceIT extends AbstractPipeDualTreeModelAutoIT {
}
}
+ @Test
+ public void testSourceAutoFlush() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> sinkAttributes = new HashMap<>();
+
+ sourceAttributes.put("source.realtime.mode", "batch");
+ sourceAttributes.put("user", "root");
+
+ sinkAttributes.put("sink", "iotdb-thrift-sink");
+ sinkAttributes.put("sink.ip", receiverIp);
+ sinkAttributes.put("sink.port", Integer.toString(receiverPort));
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db1.d1 (time, at1) values (1, 10)",
+ "insert into root.db1.d2 (time, at1) values (1, 20)"),
+ null);
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p1", sinkAttributes)
+ .setExtractorAttributes(sourceAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "insert into root.db1.d1 (time, at1) values (2, 10)",
+ "insert into root.db1.d2 (time, at1) values (2, 20)"),
+ null);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.db*.**",
+ "count(root.db1.d1.at1),count(root.db2.d1.at1),",
+ Collections.singleton("2,2,"),
+ 60);
+ }
+ }
+
@Test
public void testHistoryAndRealtime() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
index 96c519ce636..91bccc6cbda 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java
@@ -62,8 +62,7 @@ public class PipeSinkSubtask extends PipeAbstractSinkSubtask {
// to trigger the general event transfer function, causing potentially such as
// the random delay of the batch transmission. Therefore, here we inject cron events
// when no event can be pulled.
- public static final PipeHeartbeatEvent CRON_HEARTBEAT_EVENT =
- new PipeHeartbeatEvent("cron", false);
+ public static final PipeHeartbeatEvent CRON_HEARTBEAT_EVENT = new PipeHeartbeatEvent(-1, false);
public PipeSinkSubtask(
final String taskID,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java
index 3bf5c1287b7..520b57a229f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResource.java
@@ -58,15 +58,12 @@ public class DeletionResource implements PersistentResource {
private volatile Exception cause;
public DeletionResource(
- AbstractDeleteDataNode deleteDataNode,
- Consumer<DeletionResource> removeHook,
- String regionId) {
+ AbstractDeleteDataNode deleteDataNode, Consumer<DeletionResource> removeHook, int regionId) {
this.deleteDataNode = deleteDataNode;
this.removeHook = removeHook;
this.currentStatus = Status.RUNNING;
this.consensusGroupId =
- ConsensusGroupId.Factory.create(
- TConsensusGroupType.DataRegion.getValue(), Integer.parseInt(regionId));
+ ConsensusGroupId.Factory.create(TConsensusGroupType.DataRegion.getValue(), regionId);
this.pipeTaskReferenceCount =
new AtomicInteger(
DataRegionConsensusImpl.getInstance().getReplicationNum(consensusGroupId) - 1);
@@ -151,7 +148,7 @@ public class DeletionResource implements PersistentResource {
}
public static DeletionResource deserialize(
- final ByteBuffer buffer, final String regionId, final Consumer<DeletionResource> removeHook)
+ final ByteBuffer buffer, final int regionId, final Consumer<DeletionResource> removeHook)
throws IOException {
AbstractDeleteDataNode node = DeleteNodeType.deserializeFromDAL(buffer);
return new DeletionResource(node, removeHook, regionId);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
index ff868ffd445..91706fbf54f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/DeletionResourceManager.java
@@ -61,7 +61,7 @@ public class DeletionResourceManager implements AutoCloseable {
String.format(
"^_(?<%s>\\d+)-(?<%s>\\d+)\\%s$",
REBOOT_TIME, MEM_TABLE_FLUSH_ORDER, DELETION_FILE_SUFFIX);
- private final String dataRegionId;
+ private final int dataRegionId;
private final DeletionBuffer deletionBuffer;
private final File storageDir;
private final Map<AbstractDeleteDataNode, DeletionResource> deleteNode2ResourcesMap =
@@ -70,7 +70,7 @@ public class DeletionResourceManager implements AutoCloseable {
private final Condition recoveryReadyCondition = recoverLock.newCondition();
private volatile boolean hasCompletedRecovery = false;
- private DeletionResourceManager(String dataRegionId) throws IOException {
+ private DeletionResourceManager(int dataRegionId) throws IOException {
this.dataRegionId = dataRegionId;
this.storageDir =
new File(
@@ -269,7 +269,7 @@ public class DeletionResourceManager implements AutoCloseable {
//////////////////////////// singleton ////////////////////////////
private static class DeletionResourceManagerHolder {
- private static Map<String, DeletionResourceManager> CONSENSU_GROUP_ID_2_INSTANCE_MAP;
+ private static Map<Integer, DeletionResourceManager> CONSENSU_GROUP_ID_2_INSTANCE_MAP;
private DeletionResourceManagerHolder() {}
@@ -280,7 +280,7 @@ public class DeletionResourceManager implements AutoCloseable {
}
}
- public static DeletionResourceManager getInstance(String groupId) {
+ public static DeletionResourceManager getInstance(int groupId) {
// If consensusImpl is not PipeConsensus.
if (DeletionResourceManagerHolder.CONSENSU_GROUP_ID_2_INSTANCE_MAP == null) {
return null;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
index f401cbbc76a..b77105720ba 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/persist/PageCacheDeletionBuffer.java
@@ -75,7 +75,7 @@ public class PageCacheDeletionBuffer implements DeletionBuffer {
? 0
: (o1.getProgressIndex().isAfter(o2.getProgressIndex()) ? 1 : -1));
// Data region id
- private final String dataRegionId;
+ private final int dataRegionId;
// directory to store .deletion files
private final String baseDirectory;
// single thread to serialize WALEntry to workingBuffer
@@ -99,7 +99,7 @@ public class PageCacheDeletionBuffer implements DeletionBuffer {
// maxProgressIndex of each batch increases in the same order as the physical time.
private ProgressIndex maxProgressIndexInCurrentFile = MinimumProgressIndex.INSTANCE;
- public PageCacheDeletionBuffer(String dataRegionId, String baseDirectory) {
+ public PageCacheDeletionBuffer(int dataRegionId, String baseDirectory) {
this.dataRegionId = dataRegionId;
this.baseDirectory = baseDirectory;
allocateBuffers();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/recover/DeletionReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/recover/DeletionReader.java
index 943693e7b00..1fa00cd0a0e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/recover/DeletionReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/recover/DeletionReader.java
@@ -40,13 +40,13 @@ public class DeletionReader implements Closeable {
private static final Logger LOGGER = LoggerFactory.getLogger(DeletionReader.class);
private static final int MAGIC_STRING_BYTES_SIZE =
DeletionResourceManager.MAGIC_VERSION_V1.getBytes(StandardCharsets.UTF_8).length;
- private final String regionId;
+ private final int regionId;
private final Consumer<DeletionResource> removeHook;
private final File logFile;
private final FileInputStream fileInputStream;
private final FileChannel fileChannel;
- public DeletionReader(File logFile, String regionId, Consumer<DeletionResource> removeHook)
+ public DeletionReader(File logFile, int regionId, Consumer<DeletionResource> removeHook)
throws IOException {
this.logFile = logFile;
this.regionId = regionId;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index 468292b8ecc..e90ae1952aa 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -41,7 +41,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeHeartbeatEvent.class);
- private final String dataRegionId;
+ private final int dataRegionId;
private long timePublished;
private long timeAssigned;
@@ -62,7 +62,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
private final boolean shouldPrintMessage;
- public PipeHeartbeatEvent(final String dataRegionId, final boolean shouldPrintMessage) {
+ public PipeHeartbeatEvent(final int dataRegionId, final boolean shouldPrintMessage) {
super(null, 0, null, null, null, null, null, null, true, Long.MIN_VALUE, Long.MAX_VALUE);
this.dataRegionId = dataRegionId;
this.shouldPrintMessage = shouldPrintMessage;
@@ -72,7 +72,7 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
final String pipeName,
final long creationTime,
final PipeTaskMeta pipeTaskMeta,
- final String dataRegionId,
+ final int dataRegionId,
final long timePublished,
final boolean shouldPrintMessage) {
super(
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 acc62f7e7a4..540589326ff 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
@@ -57,7 +57,7 @@ public class PipeRealtimeEventFactory {
}
public static PipeRealtimeEvent createRealtimeEvent(
- final String dataRegionId, final boolean shouldPrintMessage) {
+ final int dataRegionId, final boolean shouldPrintMessage) {
return new PipeRealtimeEvent(
new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java
index 7f89cdc5372..58f8e31da26 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/source/PipeAssignerMetrics.java
@@ -41,7 +41,7 @@ public class PipeAssignerMetrics implements IMetricSet {
private AbstractMetricService metricService;
- private final Map<String, PipeDataRegionAssigner> assignerMap = new HashMap<>();
+ private final Map<Integer, PipeDataRegionAssigner> assignerMap = new HashMap<>();
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@@ -49,44 +49,44 @@ public class PipeAssignerMetrics implements IMetricSet {
public void bindTo(AbstractMetricService metricService) {
this.metricService = metricService;
synchronized (this) {
- for (String dataRegionId : assignerMap.keySet()) {
+ for (Integer dataRegionId : assignerMap.keySet()) {
createMetrics(dataRegionId);
}
}
}
- private void createMetrics(String dataRegionId) {
+ private void createMetrics(int dataRegionId) {
createAutoGauge(dataRegionId);
}
- private void createAutoGauge(String dataRegionId) {
+ private void createAutoGauge(int dataRegionId) {
metricService.createAutoGauge(
Metric.UNASSIGNED_HEARTBEAT_COUNT.toString(),
MetricLevel.IMPORTANT,
assignerMap.get(dataRegionId),
PipeDataRegionAssigner::getPipeHeartbeatEventCount,
Tag.REGION.toString(),
- dataRegionId);
+ Integer.toString(dataRegionId));
metricService.createAutoGauge(
Metric.UNASSIGNED_TABLET_COUNT.toString(),
MetricLevel.IMPORTANT,
assignerMap.get(dataRegionId),
PipeDataRegionAssigner::getTabletInsertionEventCount,
Tag.REGION.toString(),
- dataRegionId);
+ Integer.toString(dataRegionId));
metricService.createAutoGauge(
Metric.UNASSIGNED_TSFILE_COUNT.toString(),
MetricLevel.IMPORTANT,
assignerMap.get(dataRegionId),
PipeDataRegionAssigner::getTsFileInsertionEventCount,
Tag.REGION.toString(),
- dataRegionId);
+ Integer.toString(dataRegionId));
}
@Override
public void unbindFrom(AbstractMetricService metricService) {
- ImmutableSet<String> dataRegionIds = ImmutableSet.copyOf(assignerMap.keySet());
- for (String dataRegionId : dataRegionIds) {
+ ImmutableSet<Integer> dataRegionIds = ImmutableSet.copyOf(assignerMap.keySet());
+ for (int dataRegionId : dataRegionIds) {
deregister(dataRegionId);
}
if (!assignerMap.isEmpty()) {
@@ -94,32 +94,32 @@ public class PipeAssignerMetrics implements IMetricSet {
}
}
- private void removeMetrics(String dataRegionId) {
+ private void removeMetrics(int dataRegionId) {
removeAutoGauge(dataRegionId);
}
- private void removeAutoGauge(String dataRegionId) {
+ private void removeAutoGauge(int dataRegionId) {
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNASSIGNED_HEARTBEAT_COUNT.toString(),
Tag.REGION.toString(),
- dataRegionId);
+ Integer.toString(dataRegionId));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNASSIGNED_TABLET_COUNT.toString(),
Tag.REGION.toString(),
- dataRegionId);
+ Integer.toString(dataRegionId));
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.UNASSIGNED_TSFILE_COUNT.toString(),
Tag.REGION.toString(),
- dataRegionId);
+ Integer.toString(dataRegionId));
}
//////////////////////////// register & deregister (pipe integration) ////////////////////////////
public void register(PipeDataRegionAssigner pipeDataRegionAssigner) {
- String dataRegionId = pipeDataRegionAssigner.getDataRegionId();
+ int dataRegionId = pipeDataRegionAssigner.getDataRegionId();
synchronized (this) {
assignerMap.putIfAbsent(dataRegionId, pipeDataRegionAssigner);
if (Objects.nonNull(metricService)) {
@@ -128,7 +128,7 @@ public class PipeAssignerMetrics implements IMetricSet {
}
}
- public void deregister(String dataRegionId) {
+ public void deregister(int dataRegionId) {
synchronized (this) {
if (!assignerMap.containsKey(dataRegionId)) {
LOGGER.warn(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
index ab74d094740..8aac47b01c7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
@@ -498,7 +498,7 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource
extractTsFiles(dataRegion, startHistoricalExtractionTime, originalResourceList);
}
if (shouldExtractDeletion) {
- Optional.ofNullable(DeletionResourceManager.getInstance(String.valueOf(dataRegionId)))
+ Optional.ofNullable(DeletionResourceManager.getInstance(dataRegionId))
.ifPresent(manager -> extractDeletions(manager, originalResourceList));
}
@@ -946,7 +946,7 @@ public class PipeHistoricalDataRegionTsFileAndDeletionSource
dataRegionId,
event);
} else {
- Optional.ofNullable(DeletionResourceManager.getInstance(String.valueOf(dataRegionId)))
+ Optional.ofNullable(DeletionResourceManager.getInstance(dataRegionId))
.ifPresent(
manager ->
event.setDeletionResource(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index 3dec073f057..3f0d5f4687a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -91,7 +91,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
protected String pipeName;
protected long creationTime;
- protected int dataRegionId;
+ protected int dataRegionId = -1;
protected PipeTaskMeta pipeTaskMeta;
protected boolean shouldExtractInsertion;
@@ -317,7 +317,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
@Override
public void close() throws Exception {
- if (Objects.nonNull(dataRegionId)) {
+ if (dataRegionId >= 0) {
PipeInsertionDataNodeListener.getInstance().stopListenAndAssign(dataRegionId, this);
PipeTimePartitionListener.getInstance().stopListen(dataRegionId, this);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
index 8ea973fbf0a..e60e1ad898d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -66,24 +66,24 @@ public class PipeDataRegionAssigner implements Closeable {
/** The {@link DisruptorQueue} is used to assign the event to the extractor. */
private final DisruptorQueue disruptor;
- private final String dataRegionId;
+ private final int dataRegionId;
private Boolean isTableModel;
private final PipeEventCounter eventCounter = new PipeDataRegionEventCounter();
- public String getDataRegionId() {
+ public int getDataRegionId() {
return dataRegionId;
}
- public PipeDataRegionAssigner(final String dataRegionId) {
+ public PipeDataRegionAssigner(final int dataRegionId) {
this.matcher = new CachedSchemaPatternMatcher();
this.disruptor = new DisruptorQueue(this::assignToExtractor, this::onAssignedHook);
this.dataRegionId = dataRegionId;
PipeAssignerMetrics.getInstance().register(this);
final DataRegion dataRegion =
- StorageEngine.getInstance().getDataRegion(new DataRegionId(Integer.parseInt(dataRegionId)));
+ StorageEngine.getInstance().getDataRegion(new DataRegionId(dataRegionId));
if (Objects.nonNull(dataRegion)) {
final String databaseName = dataRegion.getDatabaseName();
if (Objects.nonNull(databaseName)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
index 8c49dd0a3dd..8fb95d78fc4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
@@ -48,7 +48,7 @@ import java.util.concurrent.atomic.AtomicInteger;
* will filter events and assign them to different PipeRealtimeEventDataRegionExtractors.
*/
public class PipeInsertionDataNodeListener {
- private final ConcurrentMap<String, PipeDataRegionAssigner> dataRegionId2Assigner =
+ private final ConcurrentMap<Integer, PipeDataRegionAssigner> dataRegionId2Assigner =
new ConcurrentHashMap<>();
private final AtomicInteger listenToTsFileExtractorCount = new AtomicInteger(0);
@@ -57,7 +57,7 @@ public class PipeInsertionDataNodeListener {
//////////////////////////// start & stop ////////////////////////////
public synchronized void startListenAndAssign(
- final String dataRegionId, final PipeRealtimeDataRegionSource extractor) {
+ final int dataRegionId, final PipeRealtimeDataRegionSource extractor) {
dataRegionId2Assigner
.computeIfAbsent(dataRegionId, o -> new PipeDataRegionAssigner(dataRegionId))
.startAssignTo(extractor);
@@ -71,7 +71,7 @@ public class PipeInsertionDataNodeListener {
}
public synchronized void stopListenAndAssign(
- final String dataRegionId, final PipeRealtimeDataRegionSource extractor) {
+ final int dataRegionId, final PipeRealtimeDataRegionSource extractor) {
final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
if (assigner == null) {
return;
@@ -139,7 +139,7 @@ public class PipeInsertionDataNodeListener {
}
public DeletionResource listenToDeleteData(
- final String regionId, final AbstractDeleteDataNode node) {
+ final int regionId, final AbstractDeleteDataNode node) {
final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(regionId);
// only events from registered data region will be extracted
if (assigner == null
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 d95d51e390e..8a76f891679 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
@@ -2009,7 +2009,7 @@ public class DataRegion implements IDataRegionForQuery {
}
public void deleteDALFolderAndClose() {
- Optional.ofNullable(DeletionResourceManager.getInstance(dataRegionIdString))
+ Optional.ofNullable(DeletionResourceManager.getInstance(dataRegionId.getId()))
.ifPresent(
manager -> {
manager.close();
@@ -2823,7 +2823,8 @@ public class DataRegion implements IDataRegionForQuery {
deleteDataInUnsealedFiles(unsealedTsFileResource, deletion, sealedTsFileResource);
// capture deleteDataNode and wait it to be persisted to DAL.
DeletionResource deletionResource =
- PipeInsertionDataNodeListener.getInstance().listenToDeleteData(dataRegionIdString, node);
+ PipeInsertionDataNodeListener.getInstance()
+ .listenToDeleteData(dataRegionId.getId(), node);
// just get result. We have already waited for result in `listenToDeleteData`
if (deletionResource != null && deletionResource.waitForResult() == Status.FAILURE) {
throw deletionResource.getCause();
@@ -2926,7 +2927,8 @@ public class DataRegion implements IDataRegionForQuery {
// capture deleteDataNode and wait it to be persisted to DAL.
DeletionResource deletionResource =
- PipeInsertionDataNodeListener.getInstance().listenToDeleteData(dataRegionIdString, node);
+ PipeInsertionDataNodeListener.getInstance()
+ .listenToDeleteData(dataRegionId.getId(), node);
// just get result. We have already waited for result in `listenToDeleteData`
if (deletionResource != null && deletionResource.waitForResult() == Status.FAILURE) {
throw deletionResource.getCause();
@@ -2983,7 +2985,8 @@ public class DataRegion implements IDataRegionForQuery {
deleteDataDirectlyInFile(unsealedTsFileResource, deletion);
// capture deleteDataNode and wait it to be persisted to DAL.
DeletionResource deletionResource =
- PipeInsertionDataNodeListener.getInstance().listenToDeleteData(dataRegionIdString, node);
+ PipeInsertionDataNodeListener.getInstance()
+ .listenToDeleteData(dataRegionId.getId(), node);
// just get result. We have already waited for result in `listenToDeleteData`
if (deletionResource != null && deletionResource.waitForResult() == Status.FAILURE) {
throw deletionResource.getCause();
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionRecoverTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionRecoverTest.java
index 06f823c0e23..c7a93b9d072 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionRecoverTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionRecoverTest.java
@@ -42,7 +42,7 @@ import java.io.File;
import java.util.Collections;
public class DeletionRecoverTest {
- private static final String[] FAKE_DATA_REGION_IDS = {"2", "3"};
+ private static final Integer[] FAKE_DATA_REGION_IDS = {2, 3};
private static final int THIS_DATANODE_ID =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
private static final String DELETION_BASE_DIR =
@@ -50,7 +50,7 @@ public class DeletionRecoverTest {
private final int deletionCount = 10;
private DeletionResourceManager deletionResourceManager;
- public void setUp(boolean isRelational, String FAKE_DATA_REGION_ID) throws Exception {
+ public void setUp(boolean isRelational, int FAKE_DATA_REGION_ID) throws Exception {
File baseDir = new File(DELETION_BASE_DIR + File.separator + FAKE_DATA_REGION_ID);
if (baseDir.exists()) {
FileUtils.deleteFileOrDirectory(baseDir);
@@ -84,7 +84,7 @@ public class DeletionRecoverTest {
@After
public void tearDown() throws Exception {
- for (String FAKE_DATA_REGION_ID : FAKE_DATA_REGION_IDS) {
+ for (Integer FAKE_DATA_REGION_ID : FAKE_DATA_REGION_IDS) {
File baseDir = new File(DELETION_BASE_DIR + File.separator + FAKE_DATA_REGION_ID);
if (baseDir.exists()) {
FileUtils.deleteFileOrDirectory(baseDir);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionResourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionResourceTest.java
index f94d909f94b..aa7c5c0c954 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionResourceTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/consensus/DeletionResourceTest.java
@@ -66,7 +66,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
public class DeletionResourceTest {
- private static final String[] FAKE_DATA_REGION_IDS = {"2", "3", "4", "5", "6"};
+ private static final Integer[] FAKE_DATA_REGION_IDS = {2, 3, 4, 5, 6};
private static final String DELETION_BASE_DIR =
IoTDBDescriptor.getInstance().getConfig().getIotConsensusV2DeletionFileDir();
private static final int THIS_DATANODE_ID = 0;
@@ -84,7 +84,7 @@ public class DeletionResourceTest {
@After
public void tearDown() throws Exception {
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(previousDataNodeId);
- for (String FAKE_DATA_REGION_ID : FAKE_DATA_REGION_IDS) {
+ for (Integer FAKE_DATA_REGION_ID : FAKE_DATA_REGION_IDS) {
File baseDir = new File(DELETION_BASE_DIR + File.separator + FAKE_DATA_REGION_ID);
if (baseDir.exists()) {
FileUtils.deleteFileOrDirectory(baseDir);
@@ -212,7 +212,7 @@ public class DeletionResourceTest {
new PipeDeleteDataNodeEvent(
deleteDataNode, "Test", 10, null, null, null, null, null, null, true, true);
deletionEvent.setCommitterKeyAndCommitId(
- new CommitterKey("Test", 10, Integer.parseInt(FAKE_DATA_REGION_IDS[3]), 0), i + 1);
+ new CommitterKey("Test", 10, FAKE_DATA_REGION_IDS[3], 0), i + 1);
deletionEvents.add(deletionEvent);
final DeletionResource deletionResource =
@@ -227,8 +227,7 @@ public class DeletionResourceTest {
// for event commit to invoke onCommit() to removeDAL
if (initialIndex == 0) {
- PipeEventCommitManager.getInstance()
- .register("Test", 10, Integer.parseInt(FAKE_DATA_REGION_IDS[3]), "Test");
+ PipeEventCommitManager.getInstance().register("Test", 10, FAKE_DATA_REGION_IDS[3], "Test");
}
deletionEvents.forEach(deletionEvent -> deletionEvent.increaseReferenceCount("test"));
final List<Path> paths =
@@ -255,7 +254,7 @@ public class DeletionResourceTest {
@Test
public void testWaitForResult() throws Exception {
// prepare pipe component
- final PipeRealtimeDataRegionSource extractor = new PipeRealtimeDataRegionHybridSource();
+ final PipeRealtimeDataRegionSource source = new PipeRealtimeDataRegionHybridSource();
final PipeParameters parameters =
new PipeParameters(
new HashMap<String, String>() {
@@ -265,13 +264,12 @@ public class DeletionResourceTest {
});
final PipeTaskRuntimeConfiguration configuration =
new PipeTaskRuntimeConfiguration(
- new PipeTaskSourceRuntimeEnvironment(
- "1", 1, Integer.parseInt(FAKE_DATA_REGION_IDS[4]), null));
- extractor.customize(parameters, configuration);
- Assert.assertTrue(extractor.shouldExtractDeletion());
+ new PipeTaskSourceRuntimeEnvironment("1", 1, FAKE_DATA_REGION_IDS[4], null));
+ source.customize(parameters, configuration);
+ Assert.assertTrue(source.shouldExtractDeletion());
PipeInsertionDataNodeListener.getInstance()
- .startListenAndAssign(FAKE_DATA_REGION_IDS[4], extractor);
+ .startListenAndAssign(FAKE_DATA_REGION_IDS[4], source);
deletionResourceManager = DeletionResourceManager.getInstance(FAKE_DATA_REGION_IDS[4]);
final int rebootTimes = 0;
final MeasurementPath path = new MeasurementPath("root.vehicle.d2.s0");
@@ -285,6 +283,6 @@ public class DeletionResourceTest {
Assert.assertSame(Status.SUCCESS, deletionResource.waitForResult());
// close pipe resource
PipeInsertionDataNodeListener.getInstance()
- .stopListenAndAssign(FAKE_DATA_REGION_IDS[4], extractor);
+ .stopListenAndAssign(FAKE_DATA_REGION_IDS[4], source);
}
}
(iotdb) 01/03: Flush manager
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch pipe-flush
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7bbdc95f8442ff48337f7efe3ebc45861f651535
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Feb 13 14:41:50 2026 +0800
Flush manager
---
.../tsfile/PipeCompactedTsFileInsertionEvent.java | 8 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 7 +-
.../PipeRealtimeDataRegionHybridSource.java | 13 ++-
.../realtime/PipeRealtimeDataRegionSource.java | 10 +-
.../PipeRealtimeDataRegionTsFileSource.java | 16 ++-
...ipeTsFileEpochProgressIndexAndFlushManager.java | 128 +++++++++++++++++++++
.../PipeTsFileEpochProgressIndexKeeper.java | 89 --------------
.../listener/PipeTimePartitionListener.java | 50 ++++----
.../rescon/memory/TimePartitionManager.java | 2 +-
.../apache/iotdb/commons/conf/CommonConfig.java | 15 +++
.../iotdb/commons/pipe/config/PipeConfig.java | 4 +
.../iotdb/commons/pipe/config/PipeDescriptor.java | 5 +
12 files changed, 211 insertions(+), 136 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
index bf3f2c97acb..078ce718488 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeCompactedTsFileInsertionEvent.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
-import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexAndFlushManager;
import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -38,7 +38,7 @@ import java.util.stream.Collectors;
public class PipeCompactedTsFileInsertionEvent extends PipeTsFileInsertionEvent {
- private final String dataRegionId;
+ private final int dataRegionId;
private final Set<String> originFilePaths;
private final List<Long> commitIds;
@@ -70,7 +70,7 @@ public class PipeCompactedTsFileInsertionEvent extends PipeTsFileInsertionEvent
anyOfOriginalEvents.getStartTime(),
anyOfOriginalEvents.getEndTime());
- this.dataRegionId = String.valueOf(committerKey.getRegionId());
+ this.dataRegionId = committerKey.getRegionId();
this.originFilePaths =
originalEvents.stream()
.map(PipeTsFileInsertionEvent::getTsFile)
@@ -186,7 +186,7 @@ public class PipeCompactedTsFileInsertionEvent extends PipeTsFileInsertionEvent
public void eliminateProgressIndex() {
if (Objects.isNull(overridingProgressIndex)) {
for (final String originFilePath : originFilePaths) {
- PipeTsFileEpochProgressIndexKeeper.getInstance()
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
.eliminateProgressIndex(dataRegionId, pipeName, originFilePath);
}
}
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 6b3e505d2ea..7b07e084a7a 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
@@ -45,7 +45,7 @@ import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
-import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexAndFlushManager;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -399,8 +399,9 @@ public class PipeTsFileInsertionEvent extends PipeInsertionEvent
public void eliminateProgressIndex() {
if (Objects.isNull(overridingProgressIndex) && Objects.nonNull(resource)) {
- PipeTsFileEpochProgressIndexKeeper.getInstance()
- .eliminateProgressIndex(resource.getDataRegionId(), pipeName, resource.getTsFilePath());
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
+ .eliminateProgressIndex(
+ Integer.parseInt(resource.getDataRegionId()), pipeName, resource.getTsFilePath());
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
index 0721683f4d2..bb26b5e18da 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
@@ -30,7 +30,7 @@ import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeRemainingEventAndTimeOperator;
import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
-import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexAndFlushManager;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
@@ -65,6 +65,7 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
"Unsupported event type %s for hybrid realtime extractor %s",
eventToExtract.getClass(), this));
}
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance().flushAllTimeoutTsFiles();
}
@Override
@@ -82,8 +83,8 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
if (canNotUseTabletAnymore(event)) {
event.getTsFileEpoch().migrateState(this, curState -> TsFileEpoch.State.USING_TSFILE);
- PipeTsFileEpochProgressIndexKeeper.getInstance()
- .registerProgressIndex(dataRegionId, pipeName, event.getTsFileEpoch().getResource());
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
+ .registerResource(dataRegionId, pipeName, event.getTsFileEpoch().getResource());
} else {
event
.getTsFileEpoch()
@@ -169,13 +170,15 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
switch (state) {
case USING_TABLET:
// If the state is USING_TABLET, discard the event
- PipeTsFileEpochProgressIndexKeeper.getInstance()
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
.eliminateProgressIndex(dataRegionId, pipeName, event.getTsFileEpoch().getFilePath());
event.decreaseReferenceCount(PipeRealtimeDataRegionHybridSource.class.getName(), false);
return;
case EMPTY:
case USING_TSFILE:
case USING_BOTH:
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
+ .markAsExtracted(dataRegionId, pipeName, event.getTsFileEpoch().getFilePath());
if (!pendingQueue.waitedOffer(event)) {
// This would not happen, but just in case.
// pendingQueue is unbounded, so it should never reach capacity.
@@ -310,7 +313,7 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
LOGGER.error(errorMessage);
PipeDataNodeAgent.runtime()
.report(pipeTaskMeta, new PipeRuntimeNonCriticalException(errorMessage));
- PipeTsFileEpochProgressIndexKeeper.getInstance()
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
.eliminateProgressIndex(dataRegionId, pipeName, event.getTsFileEpoch().getFilePath());
return null;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index cd00e2975a0..3dec073f057 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -39,7 +39,7 @@ import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
-import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexAndFlushManager;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeTimePartitionListener;
import org.apache.iotdb.db.storageengine.StorageEngine;
@@ -91,7 +91,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
protected String pipeName;
protected long creationTime;
- protected String dataRegionId;
+ protected int dataRegionId;
protected PipeTaskMeta pipeTaskMeta;
protected boolean shouldExtractInsertion;
@@ -214,7 +214,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
shouldExtractDeletion = insertionDeletionListeningOptionPair.getRight();
pipeName = environment.getPipeName();
- dataRegionId = String.valueOf(environment.getRegionId());
+ dataRegionId = environment.getRegionId();
pipeTaskMeta = environment.getPipeTaskMeta();
// Metrics related to TsFileEpoch are managed in PipeExtractorMetrics. These metrics are
@@ -555,7 +555,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
return skipIfNoPrivileges;
}
- public final String getDataRegionId() {
+ public final int getDataRegionId() {
return dataRegionId;
}
@@ -595,7 +595,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
}
private void maySkipProgressIndexForRealtimeEvent(final PipeRealtimeEvent event) {
- if (PipeTsFileEpochProgressIndexKeeper.getInstance()
+ if (PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
.isProgressIndexAfterOrEquals(
dataRegionId,
pipeName,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
index 8f74bc63fb3..be1f648d163 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
@@ -25,7 +25,7 @@ import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent;
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
-import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexKeeper;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeTsFileEpochProgressIndexAndFlushManager;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.epoch.TsFileEpoch;
import org.apache.iotdb.pipe.api.event.Event;
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
@@ -39,7 +39,9 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo
LoggerFactory.getLogger(PipeRealtimeDataRegionTsFileSource.class);
@Override
- protected void doExtract(PipeRealtimeEvent event) {
+ protected void doExtract(final PipeRealtimeEvent event) {
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance().flushAllTimeoutTsFiles();
+
if (event.getEvent() instanceof PipeHeartbeatEvent) {
extractHeartbeat(event);
return;
@@ -51,8 +53,8 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo
}
event.getTsFileEpoch().migrateState(this, state -> TsFileEpoch.State.USING_TSFILE);
- PipeTsFileEpochProgressIndexKeeper.getInstance()
- .registerProgressIndex(dataRegionId, pipeName, event.getTsFileEpoch().getResource());
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
+ .registerResource(dataRegionId, pipeName, event.getTsFileEpoch().getResource());
if (!(event.getEvent() instanceof TsFileInsertionEvent)) {
event.decreaseReferenceCount(PipeRealtimeDataRegionTsFileSource.class.getName(), false);
@@ -64,7 +66,7 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo
// Pending is unbounded, so it should never reach capacity.
final String errorMessage =
String.format(
- "extract: pending queue of PipeRealtimeDataRegionTsFileExtractor %s "
+ "extract: pending queue of PipeRealtimeDataRegionTsFileSource %s "
+ "has reached capacity, discard TsFile event %s, current state %s",
this, event, event.getTsFileEpoch().getState(this));
LOGGER.error(errorMessage);
@@ -75,6 +77,8 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo
event.decreaseReferenceCount(PipeRealtimeDataRegionTsFileSource.class.getName(), false);
}
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
+ .markAsExtracted(dataRegionId, pipeName, event.getTsFileEpoch().getFilePath());
event.getTsFileEpoch().clearState(this);
}
@@ -116,7 +120,7 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo
LOGGER.error(errorMessage);
PipeDataNodeAgent.runtime()
.report(pipeTaskMeta, new PipeRuntimeNonCriticalException(errorMessage));
- PipeTsFileEpochProgressIndexKeeper.getInstance()
+ PipeTsFileEpochProgressIndexAndFlushManager.getInstance()
.eliminateProgressIndex(
dataRegionId, pipeName, realtimeEvent.getTsFileEpoch().getFilePath());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexAndFlushManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexAndFlushManager.java
new file mode 100644
index 00000000000..20b2dfe410f
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexAndFlushManager.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner;
+
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.consensus.index.ProgressIndex;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.db.storageengine.StorageEngine;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+
+import org.apache.tsfile.utils.Pair;
+
+import javax.annotation.Nonnull;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class PipeTsFileEpochProgressIndexAndFlushManager {
+
+ // data region id -> pipeName -> tsFile path -> max progress index
+ private final Map<Integer, Map<String, Map<String, Pair<TsFileResource, Long>>>>
+ progressIndexKeeper = new ConcurrentHashMap<>();
+
+ public synchronized void registerResource(
+ final int dataRegionId, final String pipeName, final TsFileResource resource) {
+ progressIndexKeeper
+ .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
+ .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
+ .putIfAbsent(resource.getTsFilePath(), new Pair<>(resource, System.currentTimeMillis()));
+ }
+
+ public synchronized void markAsExtracted(
+ final int dataRegionId, final String pipeName, final String filePath) {
+ final Pair<TsFileResource, Long> pair =
+ progressIndexKeeper
+ .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
+ .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
+ .get(filePath);
+ if (Objects.nonNull(pair)) {
+ pair.setRight(Long.MAX_VALUE);
+ }
+ }
+
+ public void flushAllTimeoutTsFiles() {
+ progressIndexKeeper.forEach(
+ (regionId, map) ->
+ map.values()
+ .forEach(
+ fileMap ->
+ fileMap.forEach(
+ (path, pair) -> {
+ if (System.currentTimeMillis()
+ - PipeConfig.getInstance().getPipeTsFileFlushIntervalSeconds()
+ * 1000L
+ >= pair.getRight()) {
+ StorageEngine.getInstance()
+ .getDataRegion(new DataRegionId(regionId))
+ .asyncCloseOneTsFileProcessor(
+ pair.getLeft().isSeq(), pair.getLeft().getProcessor());
+ pair.setRight(Long.MAX_VALUE);
+ }
+ })));
+ }
+
+ public synchronized void eliminateProgressIndex(
+ final int dataRegionId, final @Nonnull String pipeName, final String filePath) {
+ progressIndexKeeper
+ .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
+ .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
+ .remove(filePath);
+ }
+
+ public synchronized boolean isProgressIndexAfterOrEquals(
+ final int dataRegionId,
+ final String pipeName,
+ final String tsFilePath,
+ final ProgressIndex progressIndex) {
+ return progressIndexKeeper
+ .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
+ .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
+ .entrySet()
+ .stream()
+ .filter(entry -> !Objects.equals(entry.getKey(), tsFilePath))
+ .map(Entry::getValue)
+ .filter(Objects::nonNull)
+ .anyMatch(resource -> !resource.getLeft().getMaxProgressIndex().isAfter(progressIndex));
+ }
+
+ //////////////////////////// singleton ////////////////////////////
+
+ private static class PipeTimePartitionProgressIndexKeeperHolder {
+
+ private static final PipeTsFileEpochProgressIndexAndFlushManager INSTANCE =
+ new PipeTsFileEpochProgressIndexAndFlushManager();
+
+ private PipeTimePartitionProgressIndexKeeperHolder() {
+ // empty constructor
+ }
+ }
+
+ public static PipeTsFileEpochProgressIndexAndFlushManager getInstance() {
+ return PipeTsFileEpochProgressIndexAndFlushManager.PipeTimePartitionProgressIndexKeeperHolder
+ .INSTANCE;
+ }
+
+ private PipeTsFileEpochProgressIndexAndFlushManager() {
+ // empty constructor
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeper.java
deleted file mode 100644
index ff7d90c377d..00000000000
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeTsFileEpochProgressIndexKeeper.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner;
-
-import org.apache.iotdb.commons.consensus.index.ProgressIndex;
-import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
-
-import javax.annotation.Nonnull;
-
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-
-public class PipeTsFileEpochProgressIndexKeeper {
-
- // data region id -> pipeName -> tsFile path -> max progress index
- private final Map<String, Map<String, Map<String, TsFileResource>>> progressIndexKeeper =
- new ConcurrentHashMap<>();
-
- public synchronized void registerProgressIndex(
- final String dataRegionId, final String pipeName, final TsFileResource resource) {
- progressIndexKeeper
- .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
- .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
- .putIfAbsent(resource.getTsFilePath(), resource);
- }
-
- public synchronized void eliminateProgressIndex(
- final String dataRegionId, final @Nonnull String pipeName, final String filePath) {
- progressIndexKeeper
- .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
- .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
- .remove(filePath);
- }
-
- public synchronized boolean isProgressIndexAfterOrEquals(
- final String dataRegionId,
- final String pipeName,
- final String tsFilePath,
- final ProgressIndex progressIndex) {
- return progressIndexKeeper
- .computeIfAbsent(dataRegionId, k -> new ConcurrentHashMap<>())
- .computeIfAbsent(pipeName, k -> new ConcurrentHashMap<>())
- .entrySet()
- .stream()
- .filter(entry -> !Objects.equals(entry.getKey(), tsFilePath))
- .map(Entry::getValue)
- .filter(Objects::nonNull)
- .anyMatch(resource -> !resource.getMaxProgressIndex().isAfter(progressIndex));
- }
-
- //////////////////////////// singleton ////////////////////////////
-
- private static class PipeTimePartitionProgressIndexKeeperHolder {
-
- private static final PipeTsFileEpochProgressIndexKeeper INSTANCE =
- new PipeTsFileEpochProgressIndexKeeper();
-
- private PipeTimePartitionProgressIndexKeeperHolder() {
- // empty constructor
- }
- }
-
- public static PipeTsFileEpochProgressIndexKeeper getInstance() {
- return PipeTsFileEpochProgressIndexKeeper.PipeTimePartitionProgressIndexKeeperHolder.INSTANCE;
- }
-
- private PipeTsFileEpochProgressIndexKeeper() {
- // empty constructor
- }
-}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java
index c5e98cd1b2c..d8fd295c9ca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeTimePartitionListener.java
@@ -30,47 +30,50 @@ import java.util.concurrent.ConcurrentHashMap;
public class PipeTimePartitionListener {
- private final Map<String, Map<String, PipeRealtimeDataRegionSource>> dataRegionId2Extractors =
+ private final Map<Integer, Map<String, PipeRealtimeDataRegionSource>> dataRegionId2Sources =
new ConcurrentHashMap<>();
// This variable is used to record the upper and lower bounds that each data region's time
// partition ID has ever reached.
- private final Map<String, Pair<Long, Long>> dataRegionId2TimePartitionIdBound =
+ private final Map<Integer, Pair<Long, Long>> dataRegionId2TimePartitionIdBound =
new ConcurrentHashMap<>();
//////////////////////////// start & stop ////////////////////////////
public synchronized void startListen(
- String dataRegionId, PipeRealtimeDataRegionSource extractor) {
- dataRegionId2Extractors
+ final int dataRegionId, final PipeRealtimeDataRegionSource source) {
+ dataRegionId2Sources
.computeIfAbsent(dataRegionId, o -> new HashMap<>())
- .put(extractor.getTaskID(), extractor);
- // Assign the previously recorded upper and lower bounds of time partition to the extractor that
+ .put(source.getTaskID(), source);
+ // Assign the previously recorded upper and lower bounds of time partition to the source that
// has just started listening to the growth of time partition.
- Pair<Long, Long> timePartitionIdBound = dataRegionId2TimePartitionIdBound.get(dataRegionId);
+ final Pair<Long, Long> timePartitionIdBound =
+ dataRegionId2TimePartitionIdBound.get(dataRegionId);
if (Objects.nonNull(timePartitionIdBound)) {
- extractor.setDataRegionTimePartitionIdBound(timePartitionIdBound);
+ source.setDataRegionTimePartitionIdBound(timePartitionIdBound);
}
}
- public synchronized void stopListen(String dataRegionId, PipeRealtimeDataRegionSource extractor) {
- Map<String, PipeRealtimeDataRegionSource> extractors =
- dataRegionId2Extractors.get(dataRegionId);
- if (Objects.isNull(extractors)) {
+ public synchronized void stopListen(
+ final int dataRegionId, final PipeRealtimeDataRegionSource source) {
+ final Map<String, PipeRealtimeDataRegionSource> sources =
+ dataRegionId2Sources.get(dataRegionId);
+ if (Objects.isNull(sources)) {
return;
}
- extractors.remove(extractor.getTaskID());
- if (extractors.isEmpty()) {
- dataRegionId2Extractors.remove(dataRegionId);
+ sources.remove(source.getTaskID());
+ if (sources.isEmpty()) {
+ dataRegionId2Sources.remove(dataRegionId);
}
}
//////////////////////////// listen to changes ////////////////////////////
public synchronized void listenToTimePartitionGrow(
- String dataRegionId, Pair<Long, Long> newTimePartitionIdBound) {
+ final int dataRegionId, final Pair<Long, Long> newTimePartitionIdBound) {
boolean shouldBroadcastTimePartitionChange = false;
- Pair<Long, Long> oldTimePartitionIdBound = dataRegionId2TimePartitionIdBound.get(dataRegionId);
+ final Pair<Long, Long> oldTimePartitionIdBound =
+ dataRegionId2TimePartitionIdBound.get(dataRegionId);
if (Objects.isNull(oldTimePartitionIdBound)) {
dataRegionId2TimePartitionIdBound.put(dataRegionId, newTimePartitionIdBound);
@@ -86,14 +89,15 @@ public class PipeTimePartitionListener {
}
if (shouldBroadcastTimePartitionChange) {
- Map<String, PipeRealtimeDataRegionSource> extractors =
- dataRegionId2Extractors.get(dataRegionId);
- if (Objects.isNull(extractors)) {
+ final Map<String, PipeRealtimeDataRegionSource> sources =
+ dataRegionId2Sources.get(dataRegionId);
+ if (Objects.isNull(sources)) {
return;
}
- Pair<Long, Long> timePartitionIdBound = dataRegionId2TimePartitionIdBound.get(dataRegionId);
- extractors.forEach(
- (id, extractor) -> extractor.setDataRegionTimePartitionIdBound(timePartitionIdBound));
+ final Pair<Long, Long> timePartitionIdBound =
+ dataRegionId2TimePartitionIdBound.get(dataRegionId);
+ sources.forEach(
+ (id, source) -> source.setDataRegionTimePartitionIdBound(timePartitionIdBound));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/TimePartitionManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/TimePartitionManager.java
index 2e1161977f8..5ded82a36ba 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/TimePartitionManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/memory/TimePartitionManager.java
@@ -67,7 +67,7 @@ public class TimePartitionManager {
// PipeInsertionDataNodeListener.listenToInsertNode.
PipeTimePartitionListener.getInstance()
.listenToTimePartitionGrow(
- String.valueOf(timePartitionInfo.dataRegionId.getId()),
+ timePartitionInfo.dataRegionId.getId(),
new Pair<>(
timePartitionInfoMapForRegion.firstKey(),
timePartitionInfoMapForRegion.lastKey()));
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index cf68da89553..2e600029241 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -322,6 +322,9 @@ public class CommonConfig {
private volatile int pipeTsFilePinMaxLogNumPerRound = 10;
private volatile int pipeTsFilePinMaxLogIntervalRounds = 90;
+ // <= 0 means disabled
+ private volatile long pipeTsFileFlushIntervalSeconds = 5 * 60L;
+
private volatile boolean pipeMemoryManagementEnabled = true;
private volatile long pipeMemoryAllocateRetryIntervalMs = 50;
private volatile int pipeMemoryAllocateMaxRetries = 10;
@@ -1783,6 +1786,18 @@ public class CommonConfig {
"pipeTsFilePinMaxLogIntervalRounds is set to {}", pipeTsFilePinMaxLogIntervalRounds);
}
+ public long getPipeTsFileFlushIntervalSeconds() {
+ return pipeTsFileFlushIntervalSeconds;
+ }
+
+ public void setPipeTsFileFlushIntervalSeconds(long pipeTsFileFlushIntervalSeconds) {
+ if (this.pipeTsFileFlushIntervalSeconds == pipeTsFileFlushIntervalSeconds) {
+ return;
+ }
+ this.pipeTsFileFlushIntervalSeconds = pipeTsFileFlushIntervalSeconds;
+ logger.info("pipeTsFileFlushIntervalSeconds is set to {}", pipeTsFileFlushIntervalSeconds);
+ }
+
public boolean getPipeMemoryManagementEnabled() {
return pipeMemoryManagementEnabled;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index a49caa53368..16539d82d8a 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -387,6 +387,10 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeTsFilePinMaxLogIntervalRounds();
}
+ public long getPipeTsFileFlushIntervalSeconds() {
+ return COMMON_CONFIG.getPipeTsFileFlushIntervalSeconds();
+ }
+
public long getPipeLoggerCacheMaxSizeInBytes() {
return COMMON_CONFIG.getPipeLoggerCacheMaxSizeInBytes();
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
index 832517b9745..6e72b4919fb 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
@@ -131,6 +131,11 @@ public class PipeDescriptor {
properties.getProperty(
"pipe_tsfile_pin_max_log_interval_rounds",
String.valueOf(config.getPipeTsFilePinMaxLogIntervalRounds()))));
+ config.setPipeTsFileFlushIntervalSeconds(
+ Long.parseLong(
+ properties.getProperty(
+ "pipe_tsfile_flush_interval_seconds",
+ String.valueOf(config.getPipeTsFileFlushIntervalSeconds()))));
config.setPipeMemoryManagementEnabled(
Boolean.parseBoolean(
properties.getProperty(