[To dev/1.3] Pipe: support path exclusion under tree model (#16632) (#16763)

* setup

* FIN

Author: VGalaxies

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

@@ -177,6 +177,80 @@ public void testPureSchemaInclusionWithMultiplePattern() throws Exception {
     }
   }
 
+  @Test
+  public void testPureSchemaInclusionWithExclusionPattern() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    final String receiverIp = receiverDataNode.getIp();
+    final int receiverPort = receiverDataNode.getPort();
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      final Map<String, String> extractorAttributes = new HashMap<>();
+      final Map<String, String> processorAttributes = new HashMap<>();
+      final Map<String, String> connectorAttributes = new HashMap<>();
+
+      extractorAttributes.put("extractor.inclusion", "schema");
+      // Include root.ln.**
+      extractorAttributes.put("path", "root.ln.**");
+      // Exclude root.ln.wf02.* and root.ln.wf03.wt01.status
+      extractorAttributes.put("path.exclusion", "root.ln.wf02.**, root.ln.wf03.wt01.status");
+
+      connectorAttributes.put("connector", "iotdb-thrift-connector");
+      connectorAttributes.put("connector.ip", receiverIp);
+      connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("testPipe", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
+
+      // Do not fail if the failure has nothing to do with pipe
+      // Because the failures will randomly generate due to resource limitation
+      if (TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              // Should be included
+              "create timeseries root.ln.wf01.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
+              "ALTER timeseries root.ln.wf01.wt01.status ADD TAGS tag3=v3",
+              "ALTER timeseries root.ln.wf01.wt01.status ADD ATTRIBUTES attr4=v4",
+              // Should be excluded by root.ln.wf02.*
+              "create timeseries root.ln.wf02.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
+              "ALTER timeseries root.ln.wf02.wt01.status ADD TAGS tag3=v3",
+              "ALTER timeseries root.ln.wf02.wt01.status ADD ATTRIBUTES attr4=v4",
+              // Should be excluded by root.ln.wf03.wt01.status
+              "create timeseries root.ln.wf03.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
+              "ALTER timeseries root.ln.wf03.wt01.status ADD TAGS tag3=v3",
+              "ALTER timeseries root.ln.wf03.wt01.status ADD ATTRIBUTES attr4=v4"))) {
+        return;
+      }
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "show timeseries root.ln.**",
+          "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
+          // Only wf01 should be synced
+          Collections.singleton(
+              "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,LZ4,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,"));
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.ln.wf01.wt01(time, status) values(now(), false)", "flush"))) {
+        return;
+      }
+
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv, "select * from root.ln.**", "Time,", Collections.emptySet());
+    }
+  }
+
   @Test
   public void testAuthExclusion() throws Exception {
     final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java

@@ -26,8 +26,8 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBPipePattern;
 import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
 import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
 import org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest;
@@ -530,14 +530,13 @@ protected TSStatus loadFileV2(
     final Set<ConfigPhysicalPlanType> executionTypes =
         PipeConfigRegionSnapshotEvent.getConfigPhysicalPlanTypeSet(
             parameters.get(ColumnHeaderConstant.TYPE));
-    final List<PipePattern> pipePatterns =
-        PipePattern.parseMultiplePatterns(
+    final PipePattern pipePattern =
+        PipePattern.parsePatternFromString(
             parameters.get(ColumnHeaderConstant.PATH_PATTERN), IoTDBPipePattern::new);
-    final PipePattern pipePattern = PipePattern.buildUnionPattern(pipePatterns);
     final List<TSStatus> results = new ArrayList<>();
     while (generator.hasNext()) {
       IoTDBConfigRegionSource.PATTERN_PARSE_VISITOR
-          .process(generator.next(), (UnionIoTDBPipePattern) pipePattern)
+          .process(generator.next(), (IoTDBPipePatternOperations) pipePattern)
           .filter(configPhysicalPlan -> executionTypes.contains(configPhysicalPlan.getType()))
           .ifPresent(
               configPhysicalPlan ->

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java

@@ -22,8 +22,8 @@
 import org.apache.iotdb.commons.auth.entity.PrivilegeType;
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBPipePattern;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanVisitor;
 import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
@@ -59,12 +59,12 @@
 
 /**
  * The {@link PipeConfigPhysicalPlanPatternParseVisitor} will transform the schema {@link
- * ConfigPhysicalPlan}s using {@link UnionIoTDBPipePattern}. Rule:
+ * ConfigPhysicalPlan}s using {@link IoTDBPipePatternOperations}. Rule:
  *
  * <p>1. All patterns in the output {@link ConfigPhysicalPlan} will be the intersection of the
- * original {@link ConfigPhysicalPlan}'s patterns and the given {@link UnionIoTDBPipePattern}.
+ * original {@link ConfigPhysicalPlan}'s patterns and the given {@link IoTDBPipePatternOperations}.
  *
- * <p>2. If a pattern does not intersect with the {@link UnionIoTDBPipePattern}, it's dropped.
+ * <p>2. If a pattern does not intersect with the {@link IoTDBPipePatternOperations}, it's dropped.
  *
  * <p>3. If all the patterns in the {@link ConfigPhysicalPlan} is dropped, the {@link
  * ConfigPhysicalPlan} is dropped.
@@ -73,13 +73,13 @@
  * one is used in the {@link PipeConfigRegionWritePlanEvent} in {@link ConfigRegionListeningQueue}.
  */
 public class PipeConfigPhysicalPlanPatternParseVisitor
-    extends ConfigPhysicalPlanVisitor<Optional<ConfigPhysicalPlan>, UnionIoTDBPipePattern> {
+    extends ConfigPhysicalPlanVisitor<Optional<ConfigPhysicalPlan>, IoTDBPipePatternOperations> {
   private static final Logger LOGGER =
       LoggerFactory.getLogger(PipeConfigPhysicalPlanPatternParseVisitor.class);
 
   @Override
   public Optional<ConfigPhysicalPlan> visitPlan(
-      final ConfigPhysicalPlan plan, final UnionIoTDBPipePattern pattern) {
+      final ConfigPhysicalPlan plan, final IoTDBPipePatternOperations pattern) {
     return Optional.of(plan);
   }
 
@@ -91,23 +91,23 @@ public Optional<ConfigPhysicalPlan> visitPlan(
   // Other matches using "matchPrefixPath" are with the same principle.
   @Override
   public Optional<ConfigPhysicalPlan> visitCreateDatabase(
-      final DatabaseSchemaPlan createDatabasePlan, final UnionIoTDBPipePattern pattern) {
+      final DatabaseSchemaPlan createDatabasePlan, final IoTDBPipePatternOperations pattern) {
     return pattern.matchPrefixPath(createDatabasePlan.getSchema().getName())
         ? Optional.of(createDatabasePlan)
         : Optional.empty();
   }
 
   @Override
   public Optional<ConfigPhysicalPlan> visitAlterDatabase(
-      final DatabaseSchemaPlan alterDatabasePlan, final UnionIoTDBPipePattern pattern) {
+      final DatabaseSchemaPlan alterDatabasePlan, final IoTDBPipePatternOperations pattern) {
     return pattern.matchPrefixPath(alterDatabasePlan.getSchema().getName())
         ? Optional.of(alterDatabasePlan)
         : Optional.empty();
   }
 
   @Override
   public Optional<ConfigPhysicalPlan> visitDeleteDatabase(
-      final DeleteDatabasePlan deleteDatabasePlan, final UnionIoTDBPipePattern pattern) {
+      final DeleteDatabasePlan deleteDatabasePlan, final IoTDBPipePatternOperations pattern) {
     return pattern.matchPrefixPath(deleteDatabasePlan.getName())
         ? Optional.of(deleteDatabasePlan)
         : Optional.empty();
@@ -116,7 +116,7 @@ public Optional<ConfigPhysicalPlan> visitDeleteDatabase(
   @Override
   public Optional<ConfigPhysicalPlan> visitCreateSchemaTemplate(
       final CreateSchemaTemplatePlan createSchemaTemplatePlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     // This is a deserialized template and can be arbitrarily altered
     final Template template = createSchemaTemplatePlan.getTemplate();
     template.getSchemaMap().keySet().removeIf(measurement -> !pattern.matchTailNode(measurement));
@@ -128,7 +128,7 @@ public Optional<ConfigPhysicalPlan> visitCreateSchemaTemplate(
   @Override
   public Optional<ConfigPhysicalPlan> visitCommitSetSchemaTemplate(
       final CommitSetSchemaTemplatePlan commitSetSchemaTemplatePlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     return pattern.matchPrefixPath(commitSetSchemaTemplatePlan.getPath())
         ? Optional.of(commitSetSchemaTemplatePlan)
         : Optional.empty();
@@ -137,7 +137,7 @@ public Optional<ConfigPhysicalPlan> visitCommitSetSchemaTemplate(
   @Override
   public Optional<ConfigPhysicalPlan> visitPipeUnsetSchemaTemplate(
       final PipeUnsetSchemaTemplatePlan pipeUnsetSchemaTemplatePlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     return pattern.matchPrefixPath(pipeUnsetSchemaTemplatePlan.getPath())
         ? Optional.of(pipeUnsetSchemaTemplatePlan)
         : Optional.empty();
@@ -146,7 +146,7 @@ public Optional<ConfigPhysicalPlan> visitPipeUnsetSchemaTemplate(
   @Override
   public Optional<ConfigPhysicalPlan> visitExtendSchemaTemplate(
       final ExtendSchemaTemplatePlan extendSchemaTemplatePlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     final TemplateExtendInfo extendInfo = extendSchemaTemplatePlan.getTemplateExtendInfo();
     final int[] filteredIndexes =
         IntStream.range(0, extendInfo.getMeasurements().size())
@@ -166,30 +166,30 @@ public Optional<ConfigPhysicalPlan> visitExtendSchemaTemplate(
 
   @Override
   public Optional<ConfigPhysicalPlan> visitGrantUser(
-      final AuthorPlan grantUserPlan, final UnionIoTDBPipePattern pattern) {
+      final AuthorPlan grantUserPlan, final IoTDBPipePatternOperations pattern) {
     return visitPathRelatedAuthorPlan(grantUserPlan, pattern);
   }
 
   @Override
   public Optional<ConfigPhysicalPlan> visitRevokeUser(
-      final AuthorPlan revokeUserPlan, final UnionIoTDBPipePattern pattern) {
+      final AuthorPlan revokeUserPlan, final IoTDBPipePatternOperations pattern) {
     return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
   }
 
   @Override
   public Optional<ConfigPhysicalPlan> visitGrantRole(
-      final AuthorPlan revokeUserPlan, final UnionIoTDBPipePattern pattern) {
+      final AuthorPlan revokeUserPlan, final IoTDBPipePatternOperations pattern) {
     return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
   }
 
   @Override
   public Optional<ConfigPhysicalPlan> visitRevokeRole(
-      final AuthorPlan revokeUserPlan, final UnionIoTDBPipePattern pattern) {
+      final AuthorPlan revokeUserPlan, final IoTDBPipePatternOperations pattern) {
     return visitPathRelatedAuthorPlan(revokeUserPlan, pattern);
   }
 
   private Optional<ConfigPhysicalPlan> visitPathRelatedAuthorPlan(
-      final AuthorPlan pathRelatedAuthorPlan, final UnionIoTDBPipePattern pattern) {
+      final AuthorPlan pathRelatedAuthorPlan, final IoTDBPipePatternOperations pattern) {
     final List<PartialPath> intersectedPaths =
         pathRelatedAuthorPlan.getNodeNameList().stream()
             .map(pattern::getIntersection)
@@ -218,7 +218,7 @@ private Optional<ConfigPhysicalPlan> visitPathRelatedAuthorPlan(
   @Override
   public Optional<ConfigPhysicalPlan> visitPipeDeleteTimeSeries(
       final PipeDeleteTimeSeriesPlan pipeDeleteTimeSeriesPlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     try {
       final PathPatternTree intersectedTree =
           pattern.getIntersection(
@@ -237,7 +237,7 @@ public Optional<ConfigPhysicalPlan> visitPipeDeleteTimeSeries(
   @Override
   public Optional<ConfigPhysicalPlan> visitPipeDeleteLogicalView(
       final PipeDeleteLogicalViewPlan pipeDeleteLogicalViewPlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     try {
       final PathPatternTree intersectedTree =
           pattern.getIntersection(
@@ -256,7 +256,7 @@ public Optional<ConfigPhysicalPlan> visitPipeDeleteLogicalView(
   @Override
   public Optional<ConfigPhysicalPlan> visitPipeDeactivateTemplate(
       final PipeDeactivateTemplatePlan pipeDeactivateTemplatePlan,
-      final UnionIoTDBPipePattern pattern) {
+      final IoTDBPipePatternOperations pattern) {
     final Map<PartialPath, List<Template>> newTemplateSetInfo =
         pipeDeactivateTemplatePlan.getTemplateSetInfo().entrySet().stream()
             .flatMap(
@@ -279,7 +279,7 @@ public Optional<ConfigPhysicalPlan> visitPipeDeactivateTemplate(
 
   @Override
   public Optional<ConfigPhysicalPlan> visitTTL(
-      final SetTTLPlan setTTLPlan, final UnionIoTDBPipePattern pattern) {
+      final SetTTLPlan setTTLPlan, final IoTDBPipePatternOperations pattern) {
     final PartialPath partialPath = new PartialPath(setTTLPlan.getPathPattern());
     final List<PartialPath> intersectionList =
         pattern.matchPrefixPath(partialPath.getFullPath())

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigPhysicalPlanPatternParseVisitor.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.path.PathPatternTree;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBPipePattern;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
 import org.apache.iotdb.confignode.consensus.request.write.auth.AuthorPlan;
@@ -55,11 +56,11 @@
 
 public class PipeConfigPhysicalPlanPatternParseVisitorTest {
 
-  private final UnionIoTDBPipePattern prefixPathPattern =
+  private final IoTDBPipePatternOperations prefixPathPattern =
       new UnionIoTDBPipePattern(new IoTDBPipePattern("root.db.device.**"));
-  private final UnionIoTDBPipePattern fullPathPattern =
+  private final IoTDBPipePatternOperations fullPathPattern =
       new UnionIoTDBPipePattern(new IoTDBPipePattern("root.db.device.s1"));
-  private final UnionIoTDBPipePattern multiplePathPattern =
+  private final IoTDBPipePatternOperations multiplePathPattern =
       new UnionIoTDBPipePattern(
           Arrays.asList(
               new IoTDBPipePattern("root.db.device.s1"),

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/source/PipeConfigPhysicalPlanPatternParseVisitorTest.java

@@ -21,7 +21,7 @@
 
 import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
 import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
 import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
 import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
@@ -158,7 +158,8 @@ private void parseAndCollectEvent(final PipeSchemaRegionWritePlanEvent deleteDat
     // delete data event does not have progress index currently
     IoTDBSchemaRegionSource.PATTERN_PARSE_VISITOR
         .process(
-            deleteDataEvent.getPlanNode(), (UnionIoTDBPipePattern) deleteDataEvent.getPipePattern())
+            deleteDataEvent.getPlanNode(),
+            (IoTDBPipePatternOperations) deleteDataEvent.getPipePattern())
         .map(
             planNode ->
                 new PipeSchemaRegionWritePlanEvent(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java

@@ -21,8 +21,8 @@
 
 import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBPipePattern;
 import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer;
 import org.apache.iotdb.db.pipe.event.common.tsfile.container.scan.TsFileInsertionScanDataContainer;
@@ -94,8 +94,8 @@ public TsFileInsertionDataContainer provide(final boolean isWithMod) throws IOEx
           isWithMod);
     }
 
-    if (pattern instanceof UnionIoTDBPipePattern
-        && !((UnionIoTDBPipePattern) pattern).mayMatchMultipleTimeSeriesInOneDevice()) {
+    if (pattern instanceof IoTDBPipePatternOperations
+        && !((IoTDBPipePatternOperations) pattern).mayMatchMultipleTimeSeriesInOneDevice()) {
       // If the pattern matches only one time series in one device, use query container here
       // because there is no timestamps merge overhead.
       //

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/container/TsFileInsertionDataContainerProvider.java

@@ -26,8 +26,8 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.pipe.config.PipeConfig;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
-import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBPipePattern;
 import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
 import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
 import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
@@ -512,10 +512,9 @@ private TSStatus loadSchemaSnapShot(
     final Set<StatementType> executionTypes =
         PipeSchemaRegionSnapshotEvent.getStatementTypeSet(
             parameters.get(ColumnHeaderConstant.TYPE));
-    final List<PipePattern> pipePatterns =
-        PipePattern.parseMultiplePatterns(
+    final PipePattern pipePattern =
+        PipePattern.parsePatternFromString(
             parameters.get(ColumnHeaderConstant.PATH_PATTERN), IoTDBPipePattern::new);
-    final PipePattern pipePattern = PipePattern.buildUnionPattern(pipePatterns);
 
     // Clear to avoid previous exceptions
     batchVisitor.clear();
@@ -530,7 +529,7 @@ private TSStatus loadSchemaSnapShot(
       // Here we apply the statements as many as possible
       // Even if there are failed statements
       STATEMENT_PATTERN_PARSE_VISITOR
-          .process(originalStatement, (UnionIoTDBPipePattern) pipePattern)
+          .process(originalStatement, (IoTDBPipePatternOperations) pipePattern)
           .flatMap(parsedStatement -> batchVisitor.process(parsedStatement, null))
           .ifPresent(statement -> results.add(executeStatementAndClassifyExceptions(statement)));
     }