Pipe: support IoTDB-style pattern

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

@@ -26,7 +26,7 @@
 import org.junit.After;
 import org.junit.Before;
 
-abstract class AbstractPipeDualIT {
+public abstract class AbstractPipeDualIT {
 
   protected BaseEnv senderEnv;
   protected BaseEnv receiverEnv;

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

@@ -130,6 +130,7 @@ public void testWithAllParameters(String realtimeMode) throws Exception {
 
       extractorAttributes.put("extractor", "iotdb-extractor");
       extractorAttributes.put("extractor.pattern", "root.db.d1");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
       extractorAttributes.put("extractor.history.enable", "true");
       extractorAttributes.put("extractor.history.start-time", "2010-01-01T08:00:00+08:00");
       extractorAttributes.put("extractor.history.end-time", "2010-01-02T08:00:00+08:00");
@@ -189,6 +190,7 @@ public void testPipeAfterDataRegionLeaderStop() throws Exception {
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", "root.db.d1");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
 
       connectorAttributes.put("connector", "iotdb-thrift-connector");
       connectorAttributes.put("connector.batch.enable", "false");
@@ -272,6 +274,7 @@ public void testPipeAfterDataRegionLeaderStop() throws Exception {
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", "root.db.d2");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
 
       connectorAttributes.put("connector", "iotdb-thrift-connector");
       connectorAttributes.put("connector.batch.enable", "false");
@@ -315,6 +318,7 @@ public void testPipeAfterRegisterNewDataNode() throws Exception {
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", "root.db.d1");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
 
       connectorAttributes.put("connector", "iotdb-thrift-connector");
       connectorAttributes.put("connector.batch.enable", "false");
@@ -368,6 +372,7 @@ public void testPipeAfterRegisterNewDataNode() throws Exception {
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", "root.db.d2");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
 
       connectorAttributes.put("connector", "iotdb-thrift-connector");
       connectorAttributes.put("connector.batch.enable", "false");

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

@@ -411,6 +411,7 @@ public void testExtractorPatternMatch() throws Exception {
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", null);
+      extractorAttributes.put("extractor.pattern.format", "prefix");
 
       connectorAttributes.put("connector", "iotdb-thrift-connector");
       connectorAttributes.put("connector.batch.enable", "false");
@@ -482,6 +483,7 @@ public void testMatchingMultipleDatabases() throws Exception {
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", "root.db1");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
 
       connectorAttributes.put("connector", "iotdb-thrift-connector");
       connectorAttributes.put("connector.batch.enable", "false");
@@ -580,6 +582,7 @@ public void testHistoryAndRealtime() throws Exception {
       connectorAttributes.put("connector.port", Integer.toString(receiverPort));
 
       extractorAttributes.put("extractor.pattern", "root.db.d2");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
       extractorAttributes.put("extractor.history.enable", "false");
       extractorAttributes.put("extractor.realtime.enable", "true");
       TSStatus status =
@@ -663,6 +666,7 @@ public void testHistoryStartTimeAndEndTimeWorkingWithOrWithoutPattern() throws E
       Map<String, String> connectorAttributes = new HashMap<>();
 
       extractorAttributes.put("extractor.pattern", "root.db.d1");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
       extractorAttributes.put("extractor.history.enable", "true");
       // 1970-01-01T08:00:02+08:00
       extractorAttributes.put("extractor.history.start-time", "2000");

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

@@ -220,6 +220,7 @@ private void testInsertNullValueTemplate(
 
       if (withParsing) {
         extractorAttributes.put("extractor.pattern", "root.sg.d1");
+        extractorAttributes.put("extractor.pattern.format", "prefix");
       }
 
       TSStatus status =

integration-test/src/test/java/org/apache/iotdb/pipe/it/pattern/PipePatternFormatIT.java

@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.it.pattern;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.MultiClusterIT2;
+import org.apache.iotdb.pipe.it.AbstractPipeDualIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({MultiClusterIT2.class})
+public class PipePatternFormatIT extends AbstractPipeDualIT {
+  @Test
+  public void testPrefixPattern() throws Exception {
+    DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    String receiverIp = receiverDataNode.getIp();
+    int receiverPort = receiverDataNode.getPort();
+
+    try (SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.db.d1(time, s, s1, t) values (1, 1, 1, 1)",
+              "insert into root.db.d2(time, s) values (1, 1)",
+              "insert into root.db2.d1(time, s) values (1, 1)"))) {
+        return;
+      }
+
+      Map<String, String> extractorAttributes = new HashMap<>();
+      Map<String, String> processorAttributes = new HashMap<>();
+      Map<String, String> connectorAttributes = new HashMap<>();
+
+      extractorAttributes.put("extractor.pattern", "root.db.d1.s");
+      extractorAttributes.put("extractor.pattern.format", "prefix");
+
+      connectorAttributes.put("connector", "iotdb-thrift-connector");
+      connectorAttributes.put("connector.batch.enable", "false");
+      connectorAttributes.put("connector.ip", receiverIp);
+      connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+      TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("p1", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+      Set<String> expectedResSet = new HashSet<>();
+      expectedResSet.add("1,1.0,1.0,");
+      TestUtils.assertDataOnEnv(
+          receiverEnv, "select * from root.**", "Time,root.db.d1.s,root.db.d1.s1,", expectedResSet);
+    }
+  }
+
+  @Test
+  public void testIotdbPattern() throws Exception {
+    DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    String receiverIp = receiverDataNode.getIp();
+    int receiverPort = receiverDataNode.getPort();
+
+    try (SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.db.d1(time, s, s1, t) values (1, 1, 1, 1)",
+              "insert into root.db.d2(time, s) values (1, 1)",
+              "insert into root.db2.d1(time, s) values (1, 1)"))) {
+        return;
+      }
+
+      Map<String, String> extractorAttributes = new HashMap<>();
+      Map<String, String> processorAttributes = new HashMap<>();
+      Map<String, String> connectorAttributes = new HashMap<>();
+
+      extractorAttributes.put("extractor.pattern", "root.**.d1.s*");
+      extractorAttributes.put("extractor.pattern.format", "iotdb");
+
+      connectorAttributes.put("connector", "iotdb-thrift-connector");
+      connectorAttributes.put("connector.batch.enable", "false");
+      connectorAttributes.put("connector.ip", receiverIp);
+      connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+      TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("p1", connectorAttributes)
+                  .setExtractorAttributes(extractorAttributes)
+                  .setProcessorAttributes(processorAttributes));
+
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
+
+      Set<String> expectedResSet = new HashSet<>();
+      expectedResSet.add("1,1.0,1.0,1.0,");
+      TestUtils.assertDataOnEnv(
+          receiverEnv,
+          "select * from root.**",
+          "Time,root.db2.d1.s,root.db.d1.s,root.db.d1.s1,",
+          expectedResSet);
+    }
+  }
+}

iotdb-connector/flink-sql-iotdb-connector/src/main/java/org/apache/iotdb/flink/sql/function/IoTDBCDCSourceFunction.java

@@ -107,6 +107,7 @@ public void open(Configuration parameters) throws Exception {
                       + "'extractor' = 'iotdb-extractor',\n"
                       + "'extractor.history.enable' = 'false',\n"
                       + "'extractor.pattern' = '%s',\n"
+                      + "'extractor.pattern.format' = 'prefix',\n"
                       + ") WITH CONNECTOR (\n"
                       + "'connector' = 'websocket-connector',\n"
                       + "'connector.websocket.port' = '%d',\n"
@@ -121,6 +122,7 @@ public void open(Configuration parameters) throws Exception {
                       + "WITH EXTRACTOR (\n"
                       + "'extractor' = 'iotdb-extractor',\n"
                       + "'extractor.pattern' = '%s',\n"
+                      + "'extractor.pattern.format' = 'prefix',\n"
                       + ") WITH CONNECTOR (\n"
                       + "'connector' = 'websocket-connector',\n"
                       + "'connector.websocket.port' = '%d',\n"

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
 import org.apache.iotdb.commons.pipe.task.meta.PipeRuntimeMeta;
 import org.apache.iotdb.commons.pipe.task.meta.PipeStaticMeta;
 import org.apache.iotdb.commons.pipe.task.meta.PipeStatus;
@@ -67,6 +68,16 @@ public CreatePipeProcedureV2() {
   public CreatePipeProcedureV2(TCreatePipeReq createPipeRequest) throws PipeException {
     super();
     this.createPipeRequest = createPipeRequest;
+    // For newly created pipes, we add an internal "source.version" attribute to identify them.
+    // Pipes without this attribute will be treated as legacy version, and their default pattern
+    // format will be "prefix".
+    // Pipes with this attribute will be treated as new version, and their default pattern format
+    // will be "iotdb".
+    this.createPipeRequest
+        .getExtractorAttributes()
+        .put(
+            PipeExtractorConstant.SOURCE_VERSION_KEY,
+            PipeExtractorConstant.SOURCE_VERSION_V2_VALUE);
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/EnrichedEvent.java

@@ -22,9 +22,10 @@
 import org.apache.iotdb.commons.consensus.index.ProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
 import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException;
-import org.apache.iotdb.commons.pipe.config.constant.PipeExtractorConstant;
 import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
 import org.apache.iotdb.db.pipe.agent.PipeAgent;
+import org.apache.iotdb.db.pipe.pattern.PipePattern;
+import org.apache.iotdb.db.pipe.pattern.PipePatternFormat;
 import org.apache.iotdb.db.pipe.progress.committer.PipeEventCommitManager;
 import org.apache.iotdb.pipe.api.event.Event;
 
@@ -50,7 +51,7 @@ public abstract class EnrichedEvent implements Event {
   public static final long NO_COMMIT_ID = -1;
   protected long commitId = NO_COMMIT_ID;
 
-  protected final String pattern;
+  protected final PipePattern pipePattern;
 
   protected final long startTime;
   protected final long endTime;
@@ -61,14 +62,18 @@ public abstract class EnrichedEvent implements Event {
   protected boolean shouldReportOnCommit = false;
 
   protected EnrichedEvent(
-      String pipeName, PipeTaskMeta pipeTaskMeta, String pattern, long startTime, long endTime) {
+      String pipeName,
+      PipeTaskMeta pipeTaskMeta,
+      PipePattern pipePattern,
+      long startTime,
+      long endTime) {
     referenceCount = new AtomicInteger(0);
     this.pipeName = pipeName;
     this.pipeTaskMeta = pipeTaskMeta;
-    this.pattern = pattern;
+    this.pipePattern = pipePattern != null ? pipePattern : new PipePattern(null);
     this.startTime = startTime;
     this.endTime = endTime;
-    isPatternParsed = getPattern().equals(PipeExtractorConstant.EXTRACTOR_PATTERN_DEFAULT_VALUE);
+    isPatternParsed = this.pipePattern.isRoot();
     isTimeParsed = Long.MIN_VALUE == startTime && Long.MAX_VALUE == endTime;
   }
 
@@ -180,12 +185,18 @@ public final PipeTaskMeta getPipeTaskMeta() {
   }
 
   /**
-   * Get the pattern of this event.
+   * Get the pattern string of this event.
    *
    * @return the pattern
    */
   public final String getPattern() {
-    return pattern == null ? PipeExtractorConstant.EXTRACTOR_PATTERN_DEFAULT_VALUE : pattern;
+    return pipePattern.getPattern() == null
+        ? pipePattern.getFormat().getDefaultPattern()
+        : pipePattern.getPattern();
+  }
+
+  public final PipePatternFormat getPatternFormat() {
+    return pipePattern.getFormat();
   }
 
   public final long getStartTime() {
@@ -221,7 +232,11 @@ public boolean shouldParseTime() {
   }
 
   public abstract EnrichedEvent shallowCopySelfAndBindPipeTaskMetaForProgressReport(
-      String pipeName, PipeTaskMeta pipeTaskMeta, String pattern, long startTime, long endTime);
+      String pipeName,
+      PipeTaskMeta pipeTaskMeta,
+      PipePattern pattern,
+      long startTime,
+      long endTime);
 
   public void reportException(PipeRuntimeException pipeRuntimeException) {
     if (pipeTaskMeta != null) {
@@ -268,7 +283,7 @@ public String toString() {
         + "', commitId="
         + commitId
         + ", pattern='"
-        + pattern
+        + pipePattern
         + "', startTime="
         + startTime
         + ", endTime="
@@ -293,7 +308,7 @@ public String coreReportMessage() {
         + "', commitId="
         + commitId
         + ", pattern='"
-        + pattern
+        + pipePattern
         + "', startTime="
         + startTime
         + ", endTime="