This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 079d2ee66de Fix snapshot pipe completion wait on unmatched DataRegions
and add IT (#18523)
079d2ee66de is described below
commit 079d2ee66de4e589fee9f783b3a083c8f95221b7
Author: Zhenyu Luo <[email protected]>
AuthorDate: Wed Aug 26 10:25:20 2026 +0800
Fix snapshot pipe completion wait on unmatched DataRegions and add IT
(#18523)
---
.../auto/enhanced/IoTDBPipeAutoDropIT.java | 107 +++++++++++++++++++++
.../runtime/heartbeat/PipeHeartbeatParser.java | 79 +++++++++++++--
2 files changed, 176 insertions(+), 10 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java
index 405c5f72371..28928653ef9 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeAutoDropIT.java
@@ -24,10 +24,14 @@ import
org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.isession.SessionConfig;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoEnhanced;
+import org.apache.iotdb.pipe.it.dual.tablemodel.TableModelUtils;
import
org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -40,9 +44,12 @@ import org.junit.runner.RunWith;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly;
@@ -130,6 +137,106 @@ public class IoTDBPipeAutoDropIT extends
AbstractPipeDualTreeModelAutoIT {
}
}
+ @Test
+ public void testAutoDropIgnoredUnmatchedDataRegions() throws Exception {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ TableModelUtils.createDataBaseAndTable(senderEnv, "t1", "table_db");
+ TableModelUtils.insertData("table_db", "t1", 0, 1, senderEnv);
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "create database root.other",
+ "insert into root.other.d1(time, s1) values (1, 1)",
+ "create database root.db",
+ "insert into root.db.d1(time, s1) values (1, 1)",
+ "flush"),
+ null);
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> sinkAttributes = new HashMap<>();
+
+ sinkAttributes.put("sink", "iotdb-thrift-sink");
+ sinkAttributes.put("sink.batch.enable", "false");
+ sinkAttributes.put("sink.ip", receiverIp);
+ sinkAttributes.put("sink.port", Integer.toString(receiverPort));
+
+ // Tree-model pipe: only listens to root.db. root.other and the table
database are
+ // user-visible DataRegions that must not block the historical snapshot
pipe from being
+ // auto-dropped.
+ final Map<String, String> treeSourceAttributes = new HashMap<>();
+ treeSourceAttributes.put("source.mode", "query");
+ treeSourceAttributes.put("source.path", "root.db.**");
+ treeSourceAttributes.put("source.capture.tree", "true");
+ treeSourceAttributes.put("source.capture.table", "false");
+ treeSourceAttributes.put("user", "root");
+
+ TSStatus status =
+ client.createPipe(
+ new TCreatePipeReq("p_tree", sinkAttributes)
+ .setExtractorAttributes(treeSourceAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe("p_tree").getCode());
+
+ // Table-model pipe: only listens to table_db, which must not be blocked
by the tree
+ // DataRegions either.
+ final Map<String, String> tableSourceAttributes = new HashMap<>();
+ tableSourceAttributes.put("source.mode", "query");
+ tableSourceAttributes.put("source.database-name", "table_db");
+ tableSourceAttributes.put("source.table-name", "t1");
+ tableSourceAttributes.put("source.capture.tree", "false");
+ tableSourceAttributes.put("source.capture.table", "true");
+ tableSourceAttributes.put("__system.sql-dialect", "table");
+ tableSourceAttributes.put("user", "root");
+
+ status =
+ client.createPipe(
+ new TCreatePipeReq("p_table", sinkAttributes)
+ .setExtractorAttributes(tableSourceAttributes)
+ .setProcessorAttributes(processorAttributes));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe("p_table").getCode());
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "select count(*) from root.db.**",
+ "count(root.db.d1.s1),",
+ Collections.singleton("1,"));
+ TableModelUtils.assertCountData("table_db", "t1", 1, receiverEnv);
+
+ await()
+ .pollInSameThread()
+ .pollDelay(1L, TimeUnit.SECONDS)
+ .pollInterval(1L, TimeUnit.SECONDS)
+ .atMost(600, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ final List<TShowPipeInfo> showPipeResult =
+ client.showPipe(new
TShowPipeReq().setUserName(SessionConfig.DEFAULT_USER))
+ .pipeInfoList;
+ showPipeResult.removeIf(
+ i ->
i.getId().startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX));
+ Assert.assertTrue(
+ showPipeResult.stream()
+ .noneMatch(
+ i ->
+ Objects.equals(i.getId(), "p_tree")
+ || Objects.equals(i.getId(), "p_table")));
+ });
+ }
+ }
+
@Test
public void testAutoDropInHistoricalTransferWithTimeRange() throws Exception
{
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
index 1149c8e4444..db3d67e86c3 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
@@ -37,6 +37,8 @@ import org.apache.iotdb.confignode.i18n.ManagerMessages;
import org.apache.iotdb.confignode.manager.ConfigManager;
import
org.apache.iotdb.confignode.manager.pipe.resource.PipeConfigNodeResourceManager;
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
+import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -169,16 +171,12 @@ public class PipeHeartbeatParser {
}
}
- final Set<Integer> requiredDataRegionIds = new HashSet<>();
- for (final Map.Entry<Integer, PipeTaskMeta> entry :
-
pipeMetaFromCoordinator.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().entrySet())
{
- if (configManager
- .getPartitionManager()
- .isRegionGroupExists(
- new TConsensusGroupId(TConsensusGroupType.DataRegion,
entry.getKey()))) {
- requiredDataRegionIds.add(entry.getKey());
- }
- }
+ // Align with copyAndFilterOutNonWorkingDataRegionPipeTasks: CN's task
table contains every
+ // user-visible DataRegion, but DataNodes only create / complete tasks
for regions that match
+ // the source pattern. Waiting on unmatched regions would prevent
snapshot pipes from
+ // dropping.
+ final Set<Integer> requiredDataRegionIds =
+ collectRequiredDataRegionIds(pipeMetaFromCoordinator);
// Remove completed pipes only when every required DataRegion has been
reported complete.
// Relying on the region-level reports (instead of the DataNode-level
boolean) prevents a
@@ -341,4 +339,65 @@ public class PipeHeartbeatParser {
}
}
}
+
+ /**
+ * Collect DataRegion ids that this pipe must wait on before auto-drop.
Schema / Config regions
+ * are skipped; DataRegions that the source pattern will not listen to are
skipped as well, using
+ * the same {@link DataRegionListeningFilter} as task push-down.
+ *
+ * <p>If database / schema lookup fails, the region is kept (same
conservative behavior as {@code
+ * copyAndFilterOutNonWorkingDataRegionPipeTasks}).
+ */
+ private Set<Integer> collectRequiredDataRegionIds(final PipeMeta
pipeMetaFromCoordinator) {
+ final PipeStaticMeta staticMeta = pipeMetaFromCoordinator.getStaticMeta();
+ final Set<Integer> requiredDataRegionIds = new HashSet<>();
+ for (final Map.Entry<Integer, PipeTaskMeta> entry :
+
pipeMetaFromCoordinator.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().entrySet())
{
+ final TConsensusGroupId dataRegionId =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion,
entry.getKey());
+ if
(!configManager.getPartitionManager().isRegionGroupExists(dataRegionId)) {
+ continue;
+ }
+ if (shouldKeepDataRegionAsRequired(staticMeta, dataRegionId)) {
+ requiredDataRegionIds.add(entry.getKey());
+ }
+ }
+ return requiredDataRegionIds;
+ }
+
+ private boolean shouldKeepDataRegionAsRequired(
+ final PipeStaticMeta staticMeta, final TConsensusGroupId dataRegionId) {
+ if (staticMeta.isSourceExternal()) {
+ return true;
+ }
+
+ final String database;
+ try {
+ database =
configManager.getPartitionManager().getRegionDatabase(dataRegionId);
+ if (database == null) {
+ return true;
+ }
+ } catch (final Exception ignored) {
+ return true;
+ }
+
+ final boolean isTableModel;
+ try {
+ final TDatabaseSchema schema =
+
configManager.getClusterSchemaManager().getDatabaseSchemaByName(database);
+ if (schema == null) {
+ return true;
+ }
+ isTableModel = schema.isIsTableModel();
+ } catch (final Exception ignored) {
+ return true;
+ }
+
+ try {
+ return DataRegionListeningFilter.shouldDatabaseBeListened(
+ staticMeta.getSourceParameters(), isTableModel, database,
staticMeta.getPipeType());
+ } catch (final Exception ignored) {
+ return true;
+ }
+ }
}
