Skip to content

Commit 8139974

Browse files
CaideyipiCaideyipi
committed
Merge remote-tracking branch 'origin/master' into fix/tag-manager-index-memory
2 parents 746f34d + cd5c3a3 commit 8139974

111 files changed

Lines changed: 3249 additions & 145 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.github/scripts/package-client-cpp-manylinux228.sh‎

100755100644
File mode changed.

‎integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,4 +174,10 @@ public DataNodeConfig setDnDataDirs(String dnDataDirs) {
174174
setProperty("dn_data_dirs", dnDataDirs);
175175
return this;
176176
}
177+
178+
@Override
179+
public DataNodeConfig setDnMultiDirStrategy(String multiDirStrategy) {
180+
setProperty("dn_multi_dir_strategy", multiDirStrategy);
181+
return this;
182+
}
177183
}

‎integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,4 +120,9 @@ public DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow) {
120120
public DataNodeConfig setDnDataDirs(String dnDataDirs) {
121121
return this;
122122
}
123+
124+
@Override
125+
public DataNodeConfig setDnMultiDirStrategy(String multiDirStrategy) {
126+
return this;
127+
}
123128
}

‎integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,4 +63,6 @@ DataNodeConfig setLoadActiveListeningCheckIntervalSeconds(
6363
DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow);
6464

6565
DataNodeConfig setDnDataDirs(String dnDataDirs);
66+
67+
DataNodeConfig setDnMultiDirStrategy(String multiDirStrategy);
6668
}

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/IoTDBRegionMigrateWithDeletionMultiDataDirIT.java‎

Lines changed: 25 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,10 @@
2020
package org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1;
2121

2222
import org.apache.iotdb.consensus.ConsensusFactory;
23-
import org.apache.iotdb.isession.SessionConfig;
2423
import org.apache.iotdb.it.env.EnvFactory;
2524
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
2625
import org.apache.iotdb.it.framework.IoTDBTestRunner;
2726
import org.apache.iotdb.itbase.category.ClusterIT;
28-
import org.apache.iotdb.itbase.env.BaseEnv;
2927

3028
import org.apache.tsfile.utils.Pair;
3129
import org.awaitility.Awaitility;
@@ -47,6 +45,13 @@
4745
import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getAllDataNodes;
4846
import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getDataRegionMapWithLeader;
4947

48+
/**
49+
* Tree-model coverage for IoTConsensus region migration over multiple data dirs: a deletion (mods)
50+
* must survive the snapshot transfer to the migrated peer. With several data dirs the snapshot
51+
* fragments of one TsFile can be received into different folders, so the receiver groups companion
52+
* files and the loader relinks them into one data dir; if that breaks, the migrated replica loses
53+
* the deletion. See the table-model twin {@link IoTDBRegionMigrateWithDeletionMultiDataDirTableIT}.
54+
*/
5055
@RunWith(IoTDBTestRunner.class)
5156
@Category({ClusterIT.class})
5257
public class IoTDBRegionMigrateWithDeletionMultiDataDirIT {
@@ -72,21 +77,20 @@ public void tearDown() throws Exception {
7277

7378
@Test
7479
public void testRegionMigratePreservesDeletionWithMultiDataDirs() throws Exception {
75-
try (Connection connection = EnvFactory.getEnv().getTableConnection();
80+
try (Connection connection = EnvFactory.getEnv().getConnection();
7681
Statement statement = connection.createStatement()) {
77-
statement.execute("CREATE DATABASE test");
78-
statement.execute("USE test");
79-
statement.execute("CREATE TABLE t1 (s1 INT64 FIELD)");
80-
statement.execute("INSERT INTO t1 (time, s1) VALUES (100, 100), (200, 200), (300, 300)");
82+
statement.execute("CREATE DATABASE root.db");
83+
statement.execute(
84+
"INSERT INTO root.db.d1(timestamp, s1) VALUES (100, 100), (200, 200), (300, 300)");
8185
statement.execute("FLUSH");
82-
statement.execute("DELETE FROM t1 WHERE time <= 200");
86+
statement.execute("DELETE FROM root.db.d1.s1 WHERE time <= 200");
8387
statement.execute("FLUSH");
8488

8589
Map<Integer, Pair<Integer, Set<Integer>>> dataRegionMapWithLeader =
8690
getDataRegionMapWithLeader(statement);
8791
int dataRegionIdForTest =
8892
dataRegionMapWithLeader.keySet().stream().max(Integer::compareTo).orElseThrow();
89-
assertDeletionVisibleOnAllReplicas(statement, dataRegionIdForTest, 1);
93+
assertDeletionVisibleOnAllReplicas(dataRegionIdForTest, 1);
9094

9195
Pair<Integer, Set<Integer>> leaderAndNodes = dataRegionMapWithLeader.get(dataRegionIdForTest);
9296
Set<Integer> allDataNodes = getAllDataNodes(statement);
@@ -123,13 +127,17 @@ public void testRegionMigratePreservesDeletionWithMultiDataDirs() throws Excepti
123127
}
124128
});
125129

126-
assertDeletionVisibleOnAllReplicas(statement, dataRegionIdForTest, 1);
130+
assertDeletionVisibleOnAllReplicas(dataRegionIdForTest, 1);
127131
}
128132
}
129133

130-
private void assertDeletionVisibleOnAllReplicas(
131-
Statement statement, int dataRegionId, int expectedCount) throws Exception {
132-
Set<Integer> replicaDataNodeIds = getReplicaDataNodeIds(statement, dataRegionId);
134+
private void assertDeletionVisibleOnAllReplicas(int dataRegionId, int expectedCount)
135+
throws Exception {
136+
Set<Integer> replicaDataNodeIds;
137+
try (Connection connection = EnvFactory.getEnv().getConnection();
138+
Statement statement = connection.createStatement()) {
139+
replicaDataNodeIds = getReplicaDataNodeIds(statement, dataRegionId);
140+
}
133141
for (int dataNodeId : replicaDataNodeIds) {
134142
DataNodeWrapper dataNodeWrapper =
135143
EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeId).orElseThrow();
@@ -143,21 +151,15 @@ private void assertDeletionVisibleOnAllReplicas(
143151

144152
private void assertDeletionVisibleOnReplica(DataNodeWrapper dataNodeWrapper, int expectedCount)
145153
throws Exception {
146-
try (Connection connection =
147-
EnvFactory.getEnv()
148-
.getConnection(
149-
dataNodeWrapper,
150-
SessionConfig.DEFAULT_USER,
151-
SessionConfig.DEFAULT_PASSWORD,
152-
BaseEnv.TABLE_SQL_DIALECT);
154+
try (Connection connection = EnvFactory.getEnv().getConnection(dataNodeWrapper);
153155
Statement dataNodeStatement = connection.createStatement()) {
154-
dataNodeStatement.execute("USE test");
155-
try (ResultSet countResultSet = dataNodeStatement.executeQuery("SELECT COUNT(s1) FROM t1")) {
156+
try (ResultSet countResultSet =
157+
dataNodeStatement.executeQuery("SELECT COUNT(s1) FROM root.db.d1")) {
156158
Assert.assertTrue(countResultSet.next());
157159
Assert.assertEquals(expectedCount, countResultSet.getLong(1));
158160
}
159161
try (ResultSet deletedRangeResultSet =
160-
dataNodeStatement.executeQuery("SELECT s1 FROM t1 WHERE time <= 200")) {
162+
dataNodeStatement.executeQuery("SELECT s1 FROM root.db.d1 WHERE time <= 200")) {
161163
Assert.assertFalse(deletedRangeResultSet.next());
162164
}
163165
}

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv1/IoTDBRegionMigrateWithDeletionMultiDataDirTableIT.java‎

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.confignode.it.regionmigration.pass.daily.iotv1;
21+
22+
import org.apache.iotdb.consensus.ConsensusFactory;
23+
import org.apache.iotdb.isession.SessionConfig;
24+
import org.apache.iotdb.it.env.EnvFactory;
25+
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
26+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
27+
import org.apache.iotdb.itbase.category.TableClusterIT;
28+
import org.apache.iotdb.itbase.env.BaseEnv;
29+
30+
import org.apache.tsfile.utils.Pair;
31+
import org.awaitility.Awaitility;
32+
import org.junit.After;
33+
import org.junit.Assert;
34+
import org.junit.Before;
35+
import org.junit.Test;
36+
import org.junit.experimental.categories.Category;
37+
import org.junit.runner.RunWith;
38+
39+
import java.sql.Connection;
40+
import java.sql.ResultSet;
41+
import java.sql.Statement;
42+
import java.util.HashSet;
43+
import java.util.Map;
44+
import java.util.Set;
45+
import java.util.concurrent.TimeUnit;
46+
47+
import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getAllDataNodes;
48+
import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getDataRegionMapWithLeader;
49+
50+
/**
51+
* Table-model twin of {@link IoTDBRegionMigrateWithDeletionMultiDataDirIT}: a deletion (mods) must
52+
* survive IoTConsensus region migration across multiple data dirs, asserted through the relational
53+
* (table) SQL dialect so the table-model cluster CI covers the same snapshot mods-transfer path.
54+
*/
55+
@RunWith(IoTDBTestRunner.class)
56+
@Category({TableClusterIT.class})
57+
public class IoTDBRegionMigrateWithDeletionMultiDataDirTableIT {
58+
59+
private static final String MULTI_DATA_DIRS =
60+
"data/datanode/data/disk0,data/datanode/data/disk1,data/datanode/data/disk2";
61+
62+
@Before
63+
public void setUp() throws Exception {
64+
EnvFactory.getEnv()
65+
.getConfig()
66+
.getCommonConfig()
67+
.setDataReplicationFactor(2)
68+
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
69+
EnvFactory.getEnv().getConfig().getDataNodeConfig().setDnDataDirs(MULTI_DATA_DIRS);
70+
EnvFactory.getEnv().initClusterEnvironment(1, 3);
71+
}
72+
73+
@After
74+
public void tearDown() throws Exception {
75+
EnvFactory.getEnv().cleanClusterEnvironment();
76+
}
77+
78+
@Test
79+
public void testRegionMigratePreservesDeletionWithMultiDataDirs() throws Exception {
80+
try (Connection connection = EnvFactory.getEnv().getTableConnection();
81+
Statement statement = connection.createStatement()) {
82+
statement.execute("CREATE DATABASE test");
83+
statement.execute("USE test");
84+
statement.execute("CREATE TABLE t1 (s1 INT64 FIELD)");
85+
statement.execute("INSERT INTO t1 (time, s1) VALUES (100, 100), (200, 200), (300, 300)");
86+
statement.execute("FLUSH");
87+
statement.execute("DELETE FROM t1 WHERE time <= 200");
88+
statement.execute("FLUSH");
89+
90+
Map<Integer, Pair<Integer, Set<Integer>>> dataRegionMapWithLeader =
91+
getDataRegionMapWithLeader(statement);
92+
int dataRegionIdForTest =
93+
dataRegionMapWithLeader.keySet().stream().max(Integer::compareTo).orElseThrow();
94+
assertDeletionVisibleOnAllReplicas(statement, dataRegionIdForTest, 1);
95+
96+
Pair<Integer, Set<Integer>> leaderAndNodes = dataRegionMapWithLeader.get(dataRegionIdForTest);
97+
Set<Integer> allDataNodes = getAllDataNodes(statement);
98+
int leaderId = leaderAndNodes.getLeft();
99+
int followerId =
100+
leaderAndNodes.getRight().stream().filter(id -> id != leaderId).findFirst().orElseThrow();
101+
int destDataNodeId =
102+
allDataNodes.stream()
103+
.filter(id -> id != leaderId && id != followerId)
104+
.findFirst()
105+
.orElseThrow();
106+
107+
statement.execute(
108+
String.format(
109+
"migrate region %d from %d to %d", dataRegionIdForTest, leaderId, destDataNodeId));
110+
111+
final int finalDestDataNodeId = destDataNodeId;
112+
Awaitility.await()
113+
.atMost(10, TimeUnit.MINUTES)
114+
.pollDelay(1, TimeUnit.SECONDS)
115+
.pollInterval(2, TimeUnit.SECONDS)
116+
.untilAsserted(
117+
() -> {
118+
try (ResultSet showRegions = statement.executeQuery("SHOW REGIONS")) {
119+
boolean migrated = false;
120+
while (showRegions.next()) {
121+
if (showRegions.getInt("RegionId") == dataRegionIdForTest
122+
&& showRegions.getInt("DataNodeId") == finalDestDataNodeId) {
123+
migrated = true;
124+
break;
125+
}
126+
}
127+
Assert.assertTrue(migrated);
128+
}
129+
});
130+
131+
assertDeletionVisibleOnAllReplicas(statement, dataRegionIdForTest, 1);
132+
}
133+
}
134+
135+
private void assertDeletionVisibleOnAllReplicas(
136+
Statement statement, int dataRegionId, int expectedCount) throws Exception {
137+
Set<Integer> replicaDataNodeIds = getReplicaDataNodeIds(statement, dataRegionId);
138+
for (int dataNodeId : replicaDataNodeIds) {
139+
DataNodeWrapper dataNodeWrapper =
140+
EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeId).orElseThrow();
141+
Awaitility.await()
142+
.atMost(2, TimeUnit.MINUTES)
143+
.pollDelay(500, TimeUnit.MILLISECONDS)
144+
.pollInterval(1, TimeUnit.SECONDS)
145+
.untilAsserted(() -> assertDeletionVisibleOnReplica(dataNodeWrapper, expectedCount));
146+
}
147+
}
148+
149+
private void assertDeletionVisibleOnReplica(DataNodeWrapper dataNodeWrapper, int expectedCount)
150+
throws Exception {
151+
try (Connection connection =
152+
EnvFactory.getEnv()
153+
.getConnection(
154+
dataNodeWrapper,
155+
SessionConfig.DEFAULT_USER,
156+
SessionConfig.DEFAULT_PASSWORD,
157+
BaseEnv.TABLE_SQL_DIALECT);
158+
Statement dataNodeStatement = connection.createStatement()) {
159+
dataNodeStatement.execute("USE test");
160+
try (ResultSet countResultSet = dataNodeStatement.executeQuery("SELECT COUNT(s1) FROM t1")) {
161+
Assert.assertTrue(countResultSet.next());
162+
Assert.assertEquals(expectedCount, countResultSet.getLong(1));
163+
}
164+
try (ResultSet deletedRangeResultSet =
165+
dataNodeStatement.executeQuery("SELECT s1 FROM t1 WHERE time <= 200")) {
166+
Assert.assertFalse(deletedRangeResultSet.next());
167+
}
168+
}
169+
}
170+
171+
private Set<Integer> getReplicaDataNodeIds(Statement statement, int dataRegionId)
172+
throws Exception {
173+
Set<Integer> replicaDataNodeIds = new HashSet<>();
174+
try (ResultSet showRegions = statement.executeQuery("SHOW REGIONS")) {
175+
while (showRegions.next()) {
176+
if ("DataRegion".equals(showRegions.getString("Type"))
177+
&& showRegions.getInt("RegionId") == dataRegionId) {
178+
replicaDataNodeIds.add(showRegions.getInt("DataNodeId"));
179+
}
180+
}
181+
}
182+
Assert.assertFalse(replicaDataNodeIds.isEmpty());
183+
return replicaDataNodeIds;
184+
}
185+
}

0 commit comments

Comments
 (0)