Skip to content

Pipe: add inclusion/exclusion pattern keys for pipe source filtering #17091

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VGalaxies wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Pipe: add inclusion/exclusion pattern keys for pipe source filtering #17091

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6797633
feat(pipe): add inclusion/exclusion pattern keys for pipe source
VGalaxies Jan 27, 2026
2f2ef2c
review
VGalaxies Jan 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
258 changes: 6 additions & 252 deletions ...est/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@

import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
Expand Down Expand Up @@ -271,68 +270,9 @@ private void testPipeWithMultiplePatterns(
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testMultiplePrefixPatternHistoricalData() throws Exception {
// Define source attributes
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.pattern", "root.db.d1.s, root.db2.d1.s");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

// Define data to be inserted
final List<String> insertQueries =
Arrays.asList(
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
"insert into root.db.d2(time, s) values (2, 2)",
"insert into root.db2.d1(time, s) values (3, 3)");

// Define expected results on receiver
final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("1,null,1.0,1.0,");
expectedResSet.add("3,3.0,null,null,");

// Execute the common test logic
testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
true, // isHistorical = true
"select * from root.db2.**,root.db.**",
"Time,root.db2.d1.s,root.db.d1.s,root.db.d1.s1,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testMultiplePrefixPatternRealtimeData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.pattern", "root.db.d1.s, root.db2.d1.s");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
"insert into root.db.d2(time, s) values (2, 2)",
"insert into root.db2.d1(time, s) values (3, 3)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("1,null,1.0,1.0,");
expectedResSet.add("3,3.0,null,null,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
false, // isHistorical = false
"select * from root.db2.**,root.db.**",
"Time,root.db2.d1.s,root.db.d1.s,root.db.d1.s1,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testMultipleIoTDBPatternHistoricalData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.path", "root.db.**, root.db2.d1.*");
sourceAttributes.put("source.pattern.inclusion", "root.db.**, root.db2.d1.*");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

Expand All @@ -358,10 +298,9 @@ public void testMultipleIoTDBPatternHistoricalData() throws Exception {
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testMultipleIoTDBPatternRealtimeData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.path", "root.db.**, root.db2.d1.*");
sourceAttributes.put("source.pattern.inclusion", "root.db.**, root.db2.d1.*");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

Expand All @@ -387,129 +326,12 @@ public void testMultipleIoTDBPatternRealtimeData() throws Exception {
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testMultipleHybridPatternHistoricalData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.path", "root.db.d1.*");
sourceAttributes.put("source.pattern", "root.db2.d1.s");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
"insert into root.db2.d1(time, s) values (2, 2)",
"insert into root.db3.d1(time, s) values (3, 3)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("1,1.0,1.0,null,");
expectedResSet.add("2,null,null,2.0,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
true, // isHistorical = true
"select * from root.db.**,root.db2.**",
"Time,root.db.d1.s,root.db.d1.s1,root.db2.d1.s,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testMultipleHybridPatternRealtimeData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.path", "root.db.d1.*");
sourceAttributes.put("source.pattern", "root.db2.d1.s");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
"insert into root.db2.d1(time, s) values (2, 2)",
"insert into root.db3.d1(time, s) values (3, 3)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("1,1.0,1.0,null,");
expectedResSet.add("2,null,null,2.0,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
false, // isHistorical = false
"select * from root.db.**,root.db2.**",
"Time,root.db.d1.s,root.db.d1.s1,root.db2.d1.s,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testPrefixPatternWithExclusionHistoricalData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
// Inclusion: Match everything under root.db.d1 and root.db.d2
sourceAttributes.put("source.pattern", "root.db.d1, root.db.d2");
// Exclusion: Exclude anything with the prefix root.db.d1.s1
sourceAttributes.put("source.pattern.exclusion", "root.db.d1.s1");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
// s matches, s1 is excluded
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
// s matches
"insert into root.db.d2(time, s) values (2, 2)",
"insert into root.db1.d1(time, s) values (3, 3)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("1,1.0,null,");
expectedResSet.add("2,null,2.0,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
true, // isHistorical = true
"select * from root.db.**",
"Time,root.db.d1.s,root.db.d2.s,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testPrefixPatternWithExclusionRealtimeData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.pattern", "root.db.d1, root.db.d2");
sourceAttributes.put("source.pattern.exclusion", "root.db.d1.s1");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
"insert into root.db.d2(time, s) values (2, 2)",
"insert into root.db1.d1(time, s) values (3, 3)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("1,1.0,null,");
expectedResSet.add("2,null,2.0,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
false, // isHistorical = false
"select * from root.db.**",
"Time,root.db.d1.s,root.db.d2.s,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testIoTDBPatternWithExclusionHistoricalData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
// Inclusion: Match everything under root.db
sourceAttributes.put("source.path", "root.db.**");
sourceAttributes.put("source.pattern.inclusion", "root.db.**");
// Exclusion: Exclude root.db.d1.s* and root.db.d3.*
sourceAttributes.put("source.path.exclusion", "root.db.d1.s*, root.db.d3.*");
sourceAttributes.put("source.pattern.exclusion", "root.db.d1.s*, root.db.d3.*");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

Expand Down Expand Up @@ -537,11 +359,10 @@ public void testIoTDBPatternWithExclusionHistoricalData() throws Exception {
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testIoTDBPatternWithExclusionRealtimeData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.path", "root.db.**");
sourceAttributes.put("source.path.exclusion", "root.db.d1.s*, root.db.d3.*");
sourceAttributes.put("source.pattern.inclusion", "root.db.**");
sourceAttributes.put("source.pattern.exclusion", "root.db.d1.s*, root.db.d3.*");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

Expand All @@ -564,71 +385,4 @@ public void testIoTDBPatternWithExclusionRealtimeData() throws Exception {
"Time,root.db.d1.t,root.db.d2.s,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testHybridPatternWithHybridExclusionHistoricalData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
// Inclusion: Match root.db.** (IoTDB) AND root.db2.d1 (Prefix)
sourceAttributes.put("source.path", "root.db.**");
sourceAttributes.put("source.pattern", "root.db2.d1");
// Exclusion: Exclude root.db.d1.* (IoTDB) AND root.db2.d1.s (Prefix)
sourceAttributes.put("source.path.exclusion", "root.db.d1.*");
sourceAttributes.put("source.pattern.exclusion", "root.db2.d1.s");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
// s, s1 excluded by path.exclusion
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
// s matches
"insert into root.db.d2(time, s) values (2, 2)",
// s excluded by pattern.exclusion, t matches
"insert into root.db2.d1(time, s, t) values (3, 3, 3)",
"insert into root.db3.d1(time, s) values (4, 4)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("2,2.0,null,");
expectedResSet.add("3,null,3.0,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
true, // isHistorical = true
"select * from root.db.**,root.db2.**",
"Time,root.db.d2.s,root.db2.d1.t,",
expectedResSet);
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testHybridPatternWithHybridExclusionRealtimeData() throws Exception {
final Map<String, String> sourceAttributes = new HashMap<>();
sourceAttributes.put("source.path", "root.db.**");
sourceAttributes.put("source.pattern", "root.db2.d1");
sourceAttributes.put("source.path.exclusion", "root.db.d1.*");
sourceAttributes.put("source.pattern.exclusion", "root.db2.d1.s");
sourceAttributes.put("source.inclusion", "data.insert");
sourceAttributes.put("user", "root");

final List<String> insertQueries =
Arrays.asList(
"insert into root.db.d1(time, s, s1) values (1, 1, 1)",
"insert into root.db.d2(time, s) values (2, 2)",
"insert into root.db2.d1(time, s, t) values (3, 3, 3)",
"insert into root.db3.d1(time, s) values (4, 4)");

final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add("2,2.0,null,");
expectedResSet.add("3,null,3.0,");

testPipeWithMultiplePatterns(
sourceAttributes,
insertQueries,
false, // isHistorical = false
"select * from root.db.**,root.db2.**",
"Time,root.db.d2.s,root.db2.d1.t,",
expectedResSet);
}
}
13 changes: 5 additions & 8 deletions ...st/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import org.apache.iotdb.rpc.TSStatusCode;

import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
Expand Down Expand Up @@ -114,7 +113,6 @@ public void testPureSchemaInclusion() throws Exception {
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testPureSchemaInclusionWithMultiplePattern() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

Expand All @@ -127,7 +125,7 @@ public void testPureSchemaInclusionWithMultiplePattern() throws Exception {
final Map<String, String> processorAttributes = new HashMap<>();
final Map<String, String> sinkAttributes = new HashMap<>();

sourceAttributes.put("path", "root.ln.wf01.wt01.status,root.ln.wf02.**");
sourceAttributes.put("source.pattern.inclusion", "root.ln.wf01.wt01.status,root.ln.wf02.**");
sourceAttributes.put("source.inclusion", "schema");
sourceAttributes.put("user", "root");

Expand Down Expand Up @@ -187,7 +185,6 @@ public void testPureSchemaInclusionWithMultiplePattern() throws Exception {
}

@Test
@Ignore("Disabled: multi/exclusion tree patterns are blocked in this branch")
public void testPureSchemaInclusionWithExclusionPattern() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

Expand All @@ -200,13 +197,13 @@ public void testPureSchemaInclusionWithExclusionPattern() throws Exception {
final Map<String, String> processorAttributes = new HashMap<>();
final Map<String, String> sinkAttributes = new HashMap<>();

sourceAttributes.put("extractor.inclusion", "schema");
sourceAttributes.put("source.inclusion", "schema");
sourceAttributes.put("user", "root");

// Include root.ln.**
sourceAttributes.put("path", "root.ln.**");
sourceAttributes.put("source.pattern.inclusion", "root.ln.**");
// Exclude root.ln.wf02.* and root.ln.wf03.wt01.status
sourceAttributes.put("path.exclusion", "root.ln.wf02.**, root.ln.wf03.wt01.status");
sourceAttributes.put("source.pattern.exclusion", "root.ln.wf02.**, root.ln.wf03.wt01.status");

sinkAttributes.put("connector", "iotdb-thrift-connector");
sinkAttributes.put("connector.ip", receiverIp);
Expand Down Expand Up @@ -314,7 +311,7 @@ public void testAuthInclusionWithPattern() throws Exception {
final Map<String, String> sinkAttributes = new HashMap<>();

sourceAttributes.put("source.inclusion", "auth");
sourceAttributes.put("path", "root.ln.**");
sourceAttributes.put("source.pattern.inclusion", "root.ln.**");
sourceAttributes.put("user", "root");

sinkAttributes.put("sink", "iotdb-thrift-sink");
Expand Down
Loading