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 ce9b3df5b39 Stabilize pipe TsFile decomposition mods IT (#18571)
ce9b3df5b39 is described below
commit ce9b3df5b3921025d5b819396b81bd8d63ac0fee
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 7 09:47:54 2026 +0800
Stabilize pipe TsFile decomposition mods IT (#18571)
---
.../IoTDBPipeTsFileDecompositionWithModsIT.java | 45 ++++++++++++++++++++++
1 file changed, 45 insertions(+)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeTsFileDecompositionWithModsIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeTsFileDecompositionWithModsIT.java
index 04e3e38167a..d010513b481 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeTsFileDecompositionWithModsIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeTsFileDecompositionWithModsIT.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.pipe.it.dual.tablemodel.manual.basic;
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.MultiClusterIT2DualTableManualBasic;
import org.apache.iotdb.itbase.env.BaseEnv;
@@ -148,6 +149,50 @@ public class IoTDBPipeTsFileDecompositionWithModsIT
extends AbstractPipeTableMod
Collections.emptySet(),
"sg1");
+ // Wait until the deletion mods are visible through every sender endpoint
before creating the
+ // pipe. Historical extraction may otherwise read a source TsFile from a
stale region leader.
+ for (DataNodeWrapper dataNode : senderEnv.getDataNodeWrapperList()) {
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ dataNode,
+ "SELECT COUNT(*) as count FROM table1 WHERE s0 ='t10' AND s1='t10'
AND s2='t10' AND s3='t10'",
+ "count,",
+ Collections.singleton("1000,"),
+ "sg2");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ dataNode,
+ "SELECT COUNT(*) as count FROM table1 WHERE s0 ='t11' AND s1='t11'
AND s2='t11' AND s3='t11'",
+ "count,",
+ Collections.singleton("0,"),
+ "sg2");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ dataNode,
+ "SELECT COUNT(*) as count FROM table1 WHERE s0 ='t12' AND s1='t12'
AND s2='t12' AND s3='t12'",
+ "count,",
+ Collections.singleton("5900,"),
+ "sg2");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ dataNode,
+ "SELECT COUNT(*) as count FROM table1 WHERE s0 ='t13' AND s1='t13'
AND s2='t13' AND s3='t13'",
+ "count,",
+ Collections.singleton("1000,"),
+ "sg2");
+
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ dataNode,
+ "SELECT COUNT(*) as count FROM table1 WHERE s0 ='t14' AND s1='t14'
AND s2='t14' AND s3='t14'",
+ "count,",
+ Collections.singleton("10000,"),
+ "sg2");
+ }
+
executeNonQueryWithRetry(
senderEnv,
String.format(
