Skip to content

Commit 41393ab

Browse files
jt2594838Caideyipi
jt2594838
and
Caideyipi
authored
Only send table deletion to related region groups (#17896)
* Only send deletion to related region group * Fix table pipe tsfile mods IT race --------- Co-authored-by: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
1 parent feaeaa6 commit 41393ab

3 files changed

Lines changed: 118 additions & 14 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeTsFileDecompositionWithModsIT.java‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -129,17 +129,31 @@ public void testTsFileDecompositionWithMods() {
129129

130130
executeNonQueryWithRetry(senderEnv, "FLUSH");
131131

132+
HashSet<String> expectedResults = new HashSet<>();
133+
expectedResults.add(
134+
"t1,t1,t1,t1,1,1.0,1,1970-01-01T00:00:00.001Z,1,1.0,1970-01-01,1,1970-01-01T00:00:00.001Z,");
135+
136+
TestUtils.assertDataEventuallyOnEnv(
137+
senderEnv,
138+
TableModelUtils.getQuerySql("table1"),
139+
TableModelUtils.generateHeaderResults(),
140+
expectedResults,
141+
"sg1");
142+
143+
TestUtils.assertDataEventuallyOnEnv(
144+
senderEnv,
145+
"SELECT s4 FROM table1 WHERE time >= 2 AND time <= 4",
146+
"s4,",
147+
Collections.emptySet(),
148+
"sg1");
149+
132150
executeNonQueryWithRetry(
133151
senderEnv,
134152
String.format(
135-
"CREATE PIPE test_pipe WITH SOURCE ('mods.enable'='true', 'capture.table'='true') WITH CONNECTOR('ip'='%s', 'port'='%s', 'username'='root', 'format'='tablet')",
153+
"CREATE PIPE test_pipe WITH SOURCE ('mods.enable'='true', 'capture.table'='true', 'inclusion'='data.insert,data.delete') WITH CONNECTOR('ip'='%s', 'port'='%s', 'username'='root', 'format'='tablet')",
136154
receiverEnv.getDataNodeWrapperList().get(0).getIp(),
137155
receiverEnv.getDataNodeWrapperList().get(0).getPort()));
138156

139-
HashSet<String> expectedResults = new HashSet<>();
140-
expectedResults.add(
141-
"t1,t1,t1,t1,1,1.0,1,1970-01-01T00:00:00.001Z,1,1.0,1970-01-01,1,1970-01-01T00:00:00.001Z,");
142-
143157
TestUtils.assertDataEventuallyOnEnv(
144158
receiverEnv,
145159
TableModelUtils.getQuerySql("table1"),

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java‎

Lines changed: 25 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
package org.apache.iotdb.db.queryengine.plan.analyze;
2121

22-
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
2322
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
2423
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
2524
import org.apache.iotdb.commons.exception.IoTDBException;
@@ -39,7 +38,8 @@
3938
import org.apache.iotdb.commons.schema.table.TsTable;
4039
import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
4140
import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
42-
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
41+
import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq;
42+
import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp;
4343
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
4444
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
4545
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
@@ -320,18 +320,34 @@ public static void analyzeDelete(final Delete node, final MPPQueryContext queryC
320320

321321
try (final ConfigNodeClient configNodeClient =
322322
ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
323-
// TODO: may use time and db/table to filter
324-
final TRegionRouteMapResp latestRegionRouteMap = configNodeClient.getLatestRegionRouteMap();
325-
final Set<TRegionReplicaSet> replicaSets = new HashSet<>();
326-
latestRegionRouteMap.getRegionRouteMap().entrySet().stream()
327-
.filter(e -> e.getKey().getType() == TConsensusGroupType.DataRegion)
328-
.forEach(e -> replicaSets.add(e.getValue()));
329-
node.setReplicaSets(replicaSets);
323+
node.setReplicaSets(fetchDeleteReplicaSets(configNodeClient, node));
324+
} catch (final IoTDBRuntimeException e) {
325+
throw e;
330326
} catch (final Exception e) {
331327
throw new IoTDBRuntimeException(e, TSStatusCode.CAN_NOT_CONNECT_CONFIGNODE.getStatusCode());
332328
}
333329
}
334330

331+
static Set<TRegionReplicaSet> fetchDeleteReplicaSets(
332+
final ConfigNodeClient configNodeClient, final Delete node) throws Exception {
333+
final Set<TRegionReplicaSet> replicaSets = new HashSet<>();
334+
for (final TableDeletionEntry tableDeletionEntry : node.getTableDeletionEntries()) {
335+
final TGetRegionGroupsByTimeResp resp =
336+
configNodeClient.getRegionGroupsByTime(
337+
new TGetRegionGroupsByTimeReq(
338+
node.getDatabaseName(),
339+
tableDeletionEntry.getStartTime(),
340+
tableDeletionEntry.getEndTime()));
341+
if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
342+
throw new IoTDBRuntimeException(resp.getStatus());
343+
}
344+
if (resp.isSetRegionReplicaSets()) {
345+
replicaSets.addAll(resp.getRegionReplicaSets());
346+
}
347+
}
348+
return replicaSets;
349+
}
350+
335351
@SuppressWarnings("java:S3655") // optional is checked
336352
public static String getDatabaseName(final Delete node, final MPPQueryContext queryContext) {
337353
final String databaseName;

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,20 +19,40 @@
1919

2020
package org.apache.iotdb.db.queryengine.plan.analyze;
2121

22+
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
23+
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
24+
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
25+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2226
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.ComparisonExpression;
2327
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
2428
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Identifier;
2529
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LongLiteral;
30+
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.QualifiedName;
31+
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Table;
2632
import org.apache.iotdb.commons.schema.table.TsTable;
2733
import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema;
34+
import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq;
35+
import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp;
36+
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
37+
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete;
38+
import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
2839
import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
40+
import org.apache.iotdb.rpc.TSStatusCode;
2941

3042
import org.apache.tsfile.enums.TSDataType;
43+
import org.apache.tsfile.read.common.TimeRange;
3144
import org.junit.Test;
45+
import org.mockito.ArgumentCaptor;
46+
import org.mockito.Mockito;
3247

48+
import java.util.Arrays;
49+
import java.util.Collections;
50+
import java.util.HashSet;
3351
import java.util.List;
52+
import java.util.Set;
3453

3554
import static org.junit.Assert.assertEquals;
55+
import static org.junit.Assert.assertTrue;
3656

3757
public class AnalyzeUtilsTest {
3858

@@ -52,4 +72,58 @@ public void testParseDeletePredicateWithRenamedTimeColumn() {
5272
assertEquals(Long.MIN_VALUE, entries.get(0).getStartTime());
5373
assertEquals(100, entries.get(0).getEndTime());
5474
}
75+
76+
@Test
77+
public void testFetchDeleteReplicaSetsOnlyQueriesTargetDatabaseRegions() throws Exception {
78+
final Delete delete = new Delete(new Table(QualifiedName.of("table1")));
79+
delete.setDatabaseName("root.db1");
80+
delete.setTableDeletionEntries(
81+
Arrays.asList(
82+
new TableDeletionEntry(new DeletionPredicate("table1"), new TimeRange(10, 20)),
83+
new TableDeletionEntry(new DeletionPredicate("table1"), new TimeRange(30, 40))));
84+
85+
final TRegionReplicaSet regionReplicaSet1 = dataRegionReplicaSet(1);
86+
final TRegionReplicaSet regionReplicaSet2 = dataRegionReplicaSet(2);
87+
final TGetRegionGroupsByTimeResp resp1 =
88+
successRegionGroupsResp(Collections.singleton(regionReplicaSet1));
89+
final TGetRegionGroupsByTimeResp resp2 =
90+
successRegionGroupsResp(new HashSet<>(Arrays.asList(regionReplicaSet1, regionReplicaSet2)));
91+
final ConfigNodeClient configNodeClient = Mockito.mock(ConfigNodeClient.class);
92+
Mockito.when(
93+
configNodeClient.getRegionGroupsByTime(Mockito.any(TGetRegionGroupsByTimeReq.class)))
94+
.thenReturn(resp1, resp2);
95+
96+
final Set<TRegionReplicaSet> result =
97+
AnalyzeUtils.fetchDeleteReplicaSets(configNodeClient, delete);
98+
99+
assertEquals(2, result.size());
100+
assertTrue(result.contains(regionReplicaSet1));
101+
assertTrue(result.contains(regionReplicaSet2));
102+
103+
final ArgumentCaptor<TGetRegionGroupsByTimeReq> reqCaptor =
104+
ArgumentCaptor.forClass(TGetRegionGroupsByTimeReq.class);
105+
Mockito.verify(configNodeClient, Mockito.times(2)).getRegionGroupsByTime(reqCaptor.capture());
106+
Mockito.verify(configNodeClient, Mockito.never()).getLatestRegionRouteMap();
107+
108+
final List<TGetRegionGroupsByTimeReq> requests = reqCaptor.getAllValues();
109+
assertEquals("root.db1", requests.get(0).getDatabase());
110+
assertEquals(10, requests.get(0).getStartTime());
111+
assertEquals(20, requests.get(0).getEndTime());
112+
assertEquals("root.db1", requests.get(1).getDatabase());
113+
assertEquals(30, requests.get(1).getStartTime());
114+
assertEquals(40, requests.get(1).getEndTime());
115+
}
116+
117+
private static TGetRegionGroupsByTimeResp successRegionGroupsResp(
118+
final Set<TRegionReplicaSet> replicaSets) {
119+
final TGetRegionGroupsByTimeResp resp =
120+
new TGetRegionGroupsByTimeResp(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
121+
resp.setRegionReplicaSets(replicaSets);
122+
return resp;
123+
}
124+
125+
private static TRegionReplicaSet dataRegionReplicaSet(final int regionId) {
126+
return new TRegionReplicaSet(
127+
new TConsensusGroupId(TConsensusGroupType.DataRegion, regionId), Collections.emptyList());
128+
}
55129
}

0 commit comments

Comments
 (0)