Skip to content

Commit 5cf809c

Browse files
VGalaxiesVGalaxies
committed
Pipe: support multiple path patterns under tree model (apache#16575)
* setup * reset * support multi for meta sync * apply review * add tests for schema sync with patterns * consider "isPrefixOrFullPath" * add more cases for PipePlanTreePatternParseVisitorTest * add more cases for PipeStatementTreePatternParseVisitorTest * add more cases for PipeConfigPhysicalPlanTreePatternParseVisitorTest * apply review * apply review * reset * reset * fix license header * fix tests * remove TODO
1 parent 2ec3441 commit 5cf809c

21 files changed

Lines changed: 1277 additions & 164 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipePatternFormatIT.java‎

Lines changed: 322 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ public void testPrefixPattern() throws Exception {
9292
}
9393

9494
@Test
95-
public void testIotdbPattern() throws Exception {
95+
public void testIoTDBPattern() throws Exception {
9696
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
9797

9898
final String receiverIp = receiverDataNode.getIp();
@@ -115,8 +115,6 @@ public void testIotdbPattern() throws Exception {
115115
final Map<String, String> connectorAttributes = new HashMap<>();
116116

117117
extractorAttributes.put("extractor.path", "root.**.d1.s*");
118-
// When path is set, pattern should be ignored
119-
extractorAttributes.put("extractor.pattern", "root");
120118
extractorAttributes.put("extractor.inclusion", "data.insert");
121119

122120
connectorAttributes.put("connector", "iotdb-thrift-connector");
@@ -146,7 +144,7 @@ public void testIotdbPattern() throws Exception {
146144
}
147145

148146
@Test
149-
public void testIotdbPatternWithLegacySyntax() throws Exception {
147+
public void testIoTDBPatternWithLegacySyntax() throws Exception {
150148
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
151149

152150
final String receiverIp = receiverDataNode.getIp();
@@ -197,4 +195,324 @@ public void testIotdbPatternWithLegacySyntax() throws Exception {
197195
expectedResSet);
198196
}
199197
}
198+
199+
@Test
200+
public void testMultiplePrefixPatternHistoricalData() throws Exception {
201+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
202+
203+
final String receiverIp = receiverDataNode.getIp();
204+
final int receiverPort = receiverDataNode.getPort();
205+
206+
try (final SyncConfigNodeIServiceClient client =
207+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
208+
final Map<String, String> extractorAttributes = new HashMap<>();
209+
final Map<String, String> processorAttributes = new HashMap<>();
210+
final Map<String, String> connectorAttributes = new HashMap<>();
211+
212+
extractorAttributes.put("extractor.pattern", "root.db.d1.s, root.db2.d1.s");
213+
extractorAttributes.put("extractor.inclusion", "data.insert");
214+
215+
connectorAttributes.put("connector", "iotdb-thrift-connector");
216+
connectorAttributes.put("connector.batch.enable", "false");
217+
connectorAttributes.put("connector.ip", receiverIp);
218+
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
219+
220+
if (!TestUtils.tryExecuteNonQueriesWithRetry(
221+
senderEnv,
222+
Arrays.asList(
223+
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
224+
"insert into root.db.d2(time, s) values (2, 2)",
225+
"insert into root.db2.d1(time, s) values (3, 3)"))) {
226+
return;
227+
}
228+
229+
final TSStatus status =
230+
client.createPipe(
231+
new TCreatePipeReq("p1", connectorAttributes)
232+
.setExtractorAttributes(extractorAttributes)
233+
.setProcessorAttributes(processorAttributes));
234+
235+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
236+
237+
Assert.assertEquals(
238+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
239+
240+
final Set<String> expectedResSet = new HashSet<>();
241+
expectedResSet.add("1,null,1.0,1.0,");
242+
expectedResSet.add("3,3.0,null,null,");
243+
TestUtils.assertDataEventuallyOnEnv(
244+
receiverEnv,
245+
"select * from root.db2.**,root.db.**",
246+
"Time,root.db2.d1.s,root.db.d1.s,root.db.d1.s1,",
247+
expectedResSet);
248+
}
249+
}
250+
251+
@Test
252+
public void testMultipleIoTDBPatternHistoricalData() throws Exception {
253+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
254+
255+
final String receiverIp = receiverDataNode.getIp();
256+
final int receiverPort = receiverDataNode.getPort();
257+
258+
try (final SyncConfigNodeIServiceClient client =
259+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
260+
final Map<String, String> extractorAttributes = new HashMap<>();
261+
final Map<String, String> processorAttributes = new HashMap<>();
262+
final Map<String, String> connectorAttributes = new HashMap<>();
263+
264+
extractorAttributes.put("extractor.path", "root.db.**, root.db2.d1.*");
265+
extractorAttributes.put("extractor.inclusion", "data.insert");
266+
267+
connectorAttributes.put("connector", "iotdb-thrift-connector");
268+
connectorAttributes.put("connector.batch.enable", "false");
269+
connectorAttributes.put("connector.ip", receiverIp);
270+
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
271+
272+
if (!TestUtils.tryExecuteNonQueriesWithRetry(
273+
senderEnv,
274+
Arrays.asList(
275+
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
276+
"insert into root.db.d2(time, s) values (2, 2)",
277+
"insert into root.db2.d1(time, s, t) values (3, 3, 3)",
278+
"insert into root.db3.d1(time, s) values (4, 4)"))) {
279+
return;
280+
}
281+
282+
final TSStatus status =
283+
client.createPipe(
284+
new TCreatePipeReq("p1", connectorAttributes)
285+
.setExtractorAttributes(extractorAttributes)
286+
.setProcessorAttributes(processorAttributes));
287+
288+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
289+
290+
Assert.assertEquals(
291+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
292+
293+
final Set<String> expectedResSet = new HashSet<>();
294+
expectedResSet.add("1,null,null,1.0,1.0,null,");
295+
expectedResSet.add("2,null,null,null,null,2.0,");
296+
expectedResSet.add("3,3.0,3.0,null,null,null,");
297+
TestUtils.assertDataEventuallyOnEnv(
298+
receiverEnv,
299+
"select * from root.db2.**,root.db.**",
300+
"Time,root.db2.d1.s,root.db2.d1.t,root.db.d1.s,root.db.d1.s1,root.db.d2.s,",
301+
expectedResSet);
302+
}
303+
}
304+
305+
@Test
306+
public void testMultipleHybridPatternHistoricalData() throws Exception {
307+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
308+
309+
final String receiverIp = receiverDataNode.getIp();
310+
final int receiverPort = receiverDataNode.getPort();
311+
312+
try (final SyncConfigNodeIServiceClient client =
313+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
314+
final Map<String, String> extractorAttributes = new HashMap<>();
315+
final Map<String, String> processorAttributes = new HashMap<>();
316+
final Map<String, String> connectorAttributes = new HashMap<>();
317+
318+
extractorAttributes.put("extractor.path", "root.db.d1.*");
319+
extractorAttributes.put("extractor.pattern", "root.db2.d1.s");
320+
extractorAttributes.put("extractor.inclusion", "data.insert");
321+
322+
connectorAttributes.put("connector", "iotdb-thrift-connector");
323+
connectorAttributes.put("connector.batch.enable", "false");
324+
connectorAttributes.put("connector.ip", receiverIp);
325+
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
326+
327+
if (!TestUtils.tryExecuteNonQueriesWithRetry(
328+
senderEnv,
329+
Arrays.asList(
330+
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
331+
"insert into root.db2.d1(time, s) values (2, 2)",
332+
"insert into root.db3.d1(time, s) values (3, 3)"))) {
333+
return;
334+
}
335+
336+
final TSStatus status =
337+
client.createPipe(
338+
new TCreatePipeReq("p1", connectorAttributes)
339+
.setExtractorAttributes(extractorAttributes)
340+
.setProcessorAttributes(processorAttributes));
341+
342+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
343+
344+
Assert.assertEquals(
345+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
346+
347+
final Set<String> expectedResSet = new HashSet<>();
348+
expectedResSet.add("1,1.0,1.0,null,");
349+
expectedResSet.add("2,null,null,2.0,");
350+
351+
TestUtils.assertDataEventuallyOnEnv(
352+
receiverEnv,
353+
"select * from root.db.**,root.db2.**",
354+
"Time,root.db.d1.s,root.db.d1.s1,root.db2.d1.s,",
355+
expectedResSet);
356+
}
357+
}
358+
359+
@Test
360+
public void testMultiplePrefixPatternRealtimeData() throws Exception {
361+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
362+
363+
final String receiverIp = receiverDataNode.getIp();
364+
final int receiverPort = receiverDataNode.getPort();
365+
366+
try (final SyncConfigNodeIServiceClient client =
367+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
368+
final Map<String, String> extractorAttributes = new HashMap<>();
369+
final Map<String, String> processorAttributes = new HashMap<>();
370+
final Map<String, String> connectorAttributes = new HashMap<>();
371+
372+
extractorAttributes.put("extractor.pattern", "root.db.d1.s, root.db2.d1.s");
373+
extractorAttributes.put("extractor.inclusion", "data.insert");
374+
375+
connectorAttributes.put("connector", "iotdb-thrift-connector");
376+
connectorAttributes.put("connector.batch.enable", "false");
377+
connectorAttributes.put("connector.ip", receiverIp);
378+
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
379+
380+
final TSStatus status =
381+
client.createPipe(
382+
new TCreatePipeReq("p1", connectorAttributes)
383+
.setExtractorAttributes(extractorAttributes)
384+
.setProcessorAttributes(processorAttributes));
385+
386+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
387+
388+
Assert.assertEquals(
389+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
390+
391+
if (!TestUtils.tryExecuteNonQueriesWithRetry(
392+
senderEnv,
393+
Arrays.asList(
394+
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
395+
"insert into root.db.d2(time, s) values (2, 2)",
396+
"insert into root.db2.d1(time, s) values (3, 3)"))) {
397+
return;
398+
}
399+
400+
final Set<String> expectedResSet = new HashSet<>();
401+
expectedResSet.add("1,null,1.0,1.0,");
402+
expectedResSet.add("3,3.0,null,null,");
403+
TestUtils.assertDataEventuallyOnEnv(
404+
receiverEnv,
405+
"select * from root.db2.**,root.db.**",
406+
"Time,root.db2.d1.s,root.db.d1.s,root.db.d1.s1,",
407+
expectedResSet);
408+
}
409+
}
410+
411+
@Test
412+
public void testMultipleIoTDBPatternRealtimeData() throws Exception {
413+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
414+
415+
final String receiverIp = receiverDataNode.getIp();
416+
final int receiverPort = receiverDataNode.getPort();
417+
418+
try (final SyncConfigNodeIServiceClient client =
419+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
420+
final Map<String, String> extractorAttributes = new HashMap<>();
421+
final Map<String, String> processorAttributes = new HashMap<>();
422+
final Map<String, String> connectorAttributes = new HashMap<>();
423+
424+
extractorAttributes.put("extractor.path", "root.db.**, root.db2.d1.*");
425+
extractorAttributes.put("extractor.inclusion", "data.insert");
426+
427+
connectorAttributes.put("connector", "iotdb-thrift-connector");
428+
connectorAttributes.put("connector.batch.enable", "false");
429+
connectorAttributes.put("connector.ip", receiverIp);
430+
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
431+
432+
final TSStatus status =
433+
client.createPipe(
434+
new TCreatePipeReq("p1", connectorAttributes)
435+
.setExtractorAttributes(extractorAttributes)
436+
.setProcessorAttributes(processorAttributes));
437+
438+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
439+
440+
Assert.assertEquals(
441+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
442+
443+
if (!TestUtils.tryExecuteNonQueriesWithRetry(
444+
senderEnv,
445+
Arrays.asList(
446+
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
447+
"insert into root.db.d2(time, s) values (2, 2)",
448+
"insert into root.db2.d1(time, s, t) values (3, 3, 3)",
449+
"insert into root.db3.d1(time, s) values (4, 4)"))) {
450+
return;
451+
}
452+
453+
final Set<String> expectedResSet = new HashSet<>();
454+
expectedResSet.add("1,null,null,1.0,1.0,null,");
455+
expectedResSet.add("2,null,null,null,null,2.0,");
456+
expectedResSet.add("3,3.0,3.0,null,null,null,");
457+
TestUtils.assertDataEventuallyOnEnv(
458+
receiverEnv,
459+
"select * from root.db2.**,root.db.**",
460+
"Time,root.db2.d1.s,root.db2.d1.t,root.db.d1.s,root.db.d1.s1,root.db.d2.s,",
461+
expectedResSet);
462+
}
463+
}
464+
465+
@Test
466+
public void testMultipleHybridPatternRealtimeData() throws Exception {
467+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
468+
469+
final String receiverIp = receiverDataNode.getIp();
470+
final int receiverPort = receiverDataNode.getPort();
471+
472+
try (final SyncConfigNodeIServiceClient client =
473+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
474+
final Map<String, String> extractorAttributes = new HashMap<>();
475+
final Map<String, String> processorAttributes = new HashMap<>();
476+
final Map<String, String> connectorAttributes = new HashMap<>();
477+
478+
extractorAttributes.put("extractor.path", "root.db.d1.*");
479+
extractorAttributes.put("extractor.pattern", "root.db2.d1.s");
480+
extractorAttributes.put("extractor.inclusion", "data.insert");
481+
482+
connectorAttributes.put("connector", "iotdb-thrift-connector");
483+
connectorAttributes.put("connector.batch.enable", "false");
484+
connectorAttributes.put("connector.ip", receiverIp);
485+
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
486+
487+
final TSStatus status =
488+
client.createPipe(
489+
new TCreatePipeReq("p1", connectorAttributes)
490+
.setExtractorAttributes(extractorAttributes)
491+
.setProcessorAttributes(processorAttributes));
492+
493+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
494+
495+
Assert.assertEquals(
496+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
497+
498+
if (!TestUtils.tryExecuteNonQueriesWithRetry(
499+
senderEnv,
500+
Arrays.asList(
501+
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
502+
"insert into root.db2.d1(time, s) values (2, 2)",
503+
"insert into root.db3.d1(time, s) values (3, 3)"))) {
504+
return;
505+
}
506+
507+
final Set<String> expectedResSet = new HashSet<>();
508+
expectedResSet.add("1,1.0,1.0,null,");
509+
expectedResSet.add("2,null,null,2.0,");
510+
511+
TestUtils.assertDataEventuallyOnEnv(
512+
receiverEnv,
513+
"select * from root.db.**,root.db2.**",
514+
"Time,root.db.d1.s,root.db.d1.s1,root.db2.d1.s,",
515+
expectedResSet);
516+
}
517+
}
200518
}

0 commit comments

Comments
 (0)