Pipe: Support "source.realtime.loose-range" = "path" in iotdb-source

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

@@ -904,7 +904,7 @@ public void testSourceStartTimeAndEndTimeWorkingWithOrWithoutPattern() throws Ex
   }
 
   @Test
-  public void testLooseRange() throws Exception {
+  public void testHistoryLooseRange() throws Exception {
     final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
 
     final String receiverIp = receiverDataNode.getIp();
@@ -964,6 +964,64 @@ public void testLooseRange() throws Exception {
     }
   }
 
+  @Test
+  public void testRealtimeLooseRange() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+    final String receiverIp = receiverDataNode.getIp();
+    final int receiverPort = receiverDataNode.getPort();
+
+    final Map<String, String> extractorAttributes = new HashMap<>();
+    final Map<String, String> processorAttributes = new HashMap<>();
+    final Map<String, String> connectorAttributes = new HashMap<>();
+
+    extractorAttributes.put("source.path", "root.db.d1.at1");
+    extractorAttributes.put("source.inclusion", "data.insert");
+    extractorAttributes.put("source.realtime.loose-range", "time, path");
+    extractorAttributes.put("source.start-time", "2000");
+    extractorAttributes.put("source.end-time", "10000");
+    extractorAttributes.put("source.realtime.mode", "batch");
+
+    connectorAttributes.put("connector", "iotdb-thrift-connector");
+    connectorAttributes.put("connector.batch.enable", "false");
+    connectorAttributes.put("connector.ip", receiverIp);
+    connectorAttributes.put("connector.port", Integer.toString(receiverPort));
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      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());
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.db.d1 (time, at1, at2)" + " values (1000, 1, 2), (3000, 3, 4)",
+              "flush"))) {
+        return;
+      }
+
+      if (!TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.db1.d1 (time, at1, at2)" + " values (1000, 1, 2), (3000, 3, 4)",
+              "flush"))) {
+        return;
+      }
+
+      TestUtils.tryExecuteNonQueriesWithRetry(
+          senderEnv,
+          Arrays.asList(
+              "insert into root.db.d1 (time, at1)" + " values (5000, 1), (16000, 3)",
+              "insert into root.db.d1 (time, at1, at2)" + " values (5001, 1, 2), (6001, 3, 4)",
+              "flush"));
+    }
+  }
+
   private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) {
     TestUtils.assertDataEventuallyOnEnv(
         receiverEnv, "count timeseries", "count(timeseries),", Collections.singleton(count + ","));

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

@@ -92,4 +92,9 @@ public boolean isGeneratedByPipe() {
   public boolean mayEventTimeOverlappedWithTimeRange() {
     return enrichedEvent.mayEventTimeOverlappedWithTimeRange();
   }
+
+  @Override
+  public boolean mayEventPathsOverlappedWithPattern() {
+    return enrichedEvent.mayEventPathsOverlappedWithPattern();
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java

@@ -121,6 +121,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() {
     return true;
   }
 
+  @Override
+  public boolean mayEventPathsOverlappedWithPattern() {
+    return true;
+  }
+
   /////////////////////////////// Whether to print ///////////////////////////////
 
   public boolean isShouldPrintMessage() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java

@@ -26,7 +26,6 @@
 import org.apache.iotdb.commons.pipe.pattern.PipePattern;
 import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
 import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
@@ -205,29 +204,72 @@ public boolean isGeneratedByPipe() {
   @Override
   public boolean mayEventTimeOverlappedWithTimeRange() {
     try {
-      final InsertNode insertNode = getInsertNode();
+      final InsertNode insertNode = getInsertNodeViaCacheIfPossible();
+      if (Objects.isNull(insertNode)) {
+        return true;
+      }
+
       if (insertNode instanceof InsertRowNode) {
         final long timestamp = ((InsertRowNode) insertNode).getTime();
         return startTime <= timestamp && timestamp <= endTime;
-      } else if (insertNode instanceof InsertTabletNode) {
+      }
+
+      if (insertNode instanceof InsertTabletNode) {
         final long[] timestamps = ((InsertTabletNode) insertNode).getTimes();
         if (Objects.isNull(timestamps) || timestamps.length == 0) {
           return false;
         }
         // We assume that `timestamps` is ordered.
         return startTime <= timestamps[timestamps.length - 1] && timestamps[0] <= endTime;
-      } else if (insertNode instanceof InsertRowsNode) {
-        for (final InsertRowNode node : ((InsertRowsNode) insertNode).getInsertRowNodeList()) {
-          final long timestamp = node.getTime();
-          if (startTime <= timestamp && timestamp <= endTime) {
-            return true;
-          }
-        }
-        return false;
-      } else {
-        throw new UnSupportedDataTypeException(
-            String.format("InsertNode type %s is not supported.", insertNode.getClass().getName()));
       }
+
+      if (insertNode instanceof InsertRowsNode) {
+        return ((InsertRowsNode) insertNode)
+            .getInsertRowNodeList().stream()
+                .anyMatch(
+                    insertRowNode -> {
+                      final long timestamp = insertRowNode.getTime();
+                      return startTime <= timestamp && timestamp <= endTime;
+                    });
+      }
+
+      return true;
+    } catch (final Exception e) {
+      LOGGER.warn(
+          "Exception occurred when determining the event time of PipeInsertNodeTabletInsertionEvent({}) overlaps with the time range: [{}, {}]. Returning true to ensure data integrity.",
+          this,
+          startTime,
+          endTime,
+          e);
+      return true;
+    }
+  }
+
+  @Override
+  public boolean mayEventPathsOverlappedWithPattern() {
+    try {
+      final InsertNode insertNode = getInsertNodeViaCacheIfPossible();
+      if (Objects.isNull(insertNode)) {
+        return true;
+      }
+
+      if (insertNode instanceof InsertRowNode || insertNode instanceof InsertTabletNode) {
+        final PartialPath devicePartialPath = insertNode.getDevicePath();
+        return Objects.isNull(devicePartialPath)
+            || pipePattern.mayOverlapWithDevice(devicePartialPath.getFullPath());
+      }
+
+      if (insertNode instanceof InsertRowsNode) {
+        return ((InsertRowsNode) insertNode)
+            .getInsertRowNodeList().stream()
+                .anyMatch(
+                    insertRowNode ->
+                        Objects.isNull(insertRowNode.getDevicePath())
+                            || pipePattern.mayOverlapWithDevice(
+                                insertRowNode.getDevicePath().getFullPath()));
+      }
+
+      return true;
     } catch (final Exception e) {
       LOGGER.warn(
           "Exception occurred when determining the event time of PipeInsertNodeTabletInsertionEvent({}) overlaps with the time range: [{}, {}]. Returning true to ensure data integrity.",
@@ -318,22 +360,6 @@ public long count() {
 
   /////////////////////////// parsePatternOrTime ///////////////////////////
 
-  @Override
-  public boolean shouldParsePattern() {
-    final InsertNode node = getInsertNodeViaCacheIfPossible();
-    return super.shouldParsePattern()
-        && Objects.nonNull(pipePattern)
-        && (Objects.isNull(node)
-            || (node.getType() == PlanNodeType.INSERT_ROWS
-                ? ((InsertRowsNode) node)
-                    .getInsertRowNodeList().stream()
-                        .anyMatch(
-                            insertRowNode ->
-                                !pipePattern.coversDevice(
-                                    insertRowNode.getDevicePath().getFullPath()))
-                : !pipePattern.coversDevice(node.getDevicePath().getFullPath())));
-  }
-
   public List<PipeRawTabletInsertionEvent> toRawTabletInsertionEvents() {
     final List<PipeRawTabletInsertionEvent> events =
         convertToTablets().stream()

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java

@@ -191,6 +191,12 @@ public boolean mayEventTimeOverlappedWithTimeRange() {
     return startTime <= timestamps[timestamps.length - 1] && timestamps[0] <= endTime;
   }
 
+  @Override
+  public boolean mayEventPathsOverlappedWithPattern() {
+    final String deviceId = getDeviceId();
+    return Objects.isNull(deviceId) || pipePattern.mayOverlapWithDevice(deviceId);
+  }
+
   public void markAsNeedToReport() {
     this.needToReport = true;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java

@@ -84,6 +84,11 @@ public boolean mayEventTimeOverlappedWithTimeRange() {
     return true;
   }
 
+  @Override
+  public boolean mayEventPathsOverlappedWithPattern() {
+    return true;
+  }
+
   @Override
   public void reportProgress() {
     PipeAgent.task().markCompleted(pipeName, dataRegionId);

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

@@ -26,19 +26,25 @@
 import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.resource.PipeResourceManager;
+import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.PlainDeviceID;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class PipeTsFileInsertionEvent extends EnrichedEvent implements TsFileInsertionEvent {
@@ -277,28 +283,35 @@ public boolean mayEventTimeOverlappedWithTimeRange() {
         : resource.getFileStartTime() <= endTime;
   }
 
-  /////////////////////////// TsFileInsertionEvent ///////////////////////////
-
   @Override
-  public boolean shouldParseTimeOrPattern() {
-    boolean shouldParseTimeOrPattern = false;
+  public boolean mayEventPathsOverlappedWithPattern() {
+    if (!resource.isClosed()) {
+      return true;
+    }
+
     try {
-      shouldParseTimeOrPattern = super.shouldParseTimeOrPattern();
-      return shouldParseTimeOrPattern;
-    } finally {
-      // Super method will call shouldParsePattern() and then init dataContainer at
-      // shouldParsePattern(). If shouldParsePattern() returns false, dataContainer will
-      // not be used, so we need to close the resource here.
-      if (!shouldParseTimeOrPattern) {
-        close();
-      }
+      final Map<IDeviceID, Boolean> deviceIsAlignedMap =
+          PipeResourceManager.tsfile()
+              .getDeviceIsAlignedMapFromCache(
+                  PipeTsFileResourceManager.getHardlinkOrCopiedFileInPipeDir(resource.getTsFile()));
+      final Set<IDeviceID> deviceSet =
+          Objects.nonNull(deviceIsAlignedMap) ? deviceIsAlignedMap.keySet() : resource.getDevices();
+      return deviceSet.stream()
+          .anyMatch(
+              // TODO: use IDeviceID
+              deviceID ->
+                  pipePattern.mayOverlapWithDevice(((PlainDeviceID) deviceID).toStringID()));
+    } catch (final IOException e) {
+      LOGGER.warn(
+          "Pipe {}: failed to get devices from TsFile {}, extract it anyway",
+          pipeName,
+          resource.getTsFilePath(),
+          e);
+      return true;
     }
   }
 
-  @Override
-  public boolean shouldParsePattern() {
-    return super.shouldParsePattern() && initDataContainer().shouldParsePattern();
-  }
+  /////////////////////////// TsFileInsertionEvent ///////////////////////////
 
   @Override
   public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {

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

@@ -129,10 +129,6 @@ public ProgressIndex getProgressIndex() {
     return event.getProgressIndex();
   }
 
-  /**
-   * If pipe's pattern is database-level, then no need to parse event by pattern cause pipes are
-   * data-region-level.
-   */
   @Override
   public void skipParsingPattern() {
     event.skipParsingPattern();
@@ -143,6 +139,26 @@ public void skipParsingTime() {
     event.skipParsingTime();
   }
 
+  @Override
+  public boolean shouldParseTime() {
+    return event.shouldParseTime();
+  }
+
+  @Override
+  public boolean shouldParsePattern() {
+    return event.shouldParsePattern();
+  }
+
+  @Override
+  public boolean mayEventTimeOverlappedWithTimeRange() {
+    return event.mayEventTimeOverlappedWithTimeRange();
+  }
+
+  @Override
+  public boolean mayEventPathsOverlappedWithPattern() {
+    return event.mayEventPathsOverlappedWithPattern();
+  }
+
   @Override
   public PipeRealtimeEvent shallowCopySelfAndBindPipeTaskMetaForProgressReport(
       final String pipeName,
@@ -167,11 +183,6 @@ public boolean isGeneratedByPipe() {
     return event.isGeneratedByPipe();
   }
 
-  @Override
-  public boolean mayEventTimeOverlappedWithTimeRange() {
-    return event.mayEventTimeOverlappedWithTimeRange();
-  }
-
   @Override
   public String toString() {
     return "PipeRealtimeEvent{"

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileExtractor.java

@@ -109,8 +109,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa
   private long historicalDataExtractionEndTime = Long.MAX_VALUE; // Event time
   private long historicalDataExtractionTimeLowerBound; // Arrival time
 
-  private boolean sloppyPattern;
   private boolean sloppyTimeRange; // true to disable time range filter after extraction
+  private boolean sloppyPattern; // true to disable pattern filter after extraction
 
   private Pair<Boolean, Boolean> listeningOptionPair;
   private boolean shouldExtractInsertion;
@@ -142,10 +142,9 @@ public void validate(final PipeParameterValidator validator) {
                         EXTRACTOR_HISTORY_LOOSE_RANGE_DEFAULT_VALUE)
                     .split(","))
             .map(String::trim)
+            .filter(s -> !s.isEmpty())
             .map(String::toLowerCase)
             .collect(Collectors.toSet());
-    // Avoid empty string
-    sloppyOptionSet.remove("");
     sloppyTimeRange = sloppyOptionSet.remove(EXTRACTOR_HISTORY_LOOSE_RANGE_TIME_VALUE);
     sloppyPattern = sloppyOptionSet.remove(EXTRACTOR_HISTORY_LOOSE_RANGE_PATH_VALUE);
     if (!sloppyOptionSet.isEmpty()) {
@@ -183,6 +182,8 @@ public void validate(final PipeParameterValidator validator) {
                   EXTRACTOR_END_TIME_KEY,
                   historicalDataExtractionEndTime));
         }
+      } catch (final PipeParameterNotValidException e) {
+        throw e;
       } catch (final Exception e) {
         // compatible with the current validation framework
         throw new PipeParameterNotValidException(e.getMessage());