[To dev/1.3] [Pipe] Respect history enable with source time range (#18362)

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

@@ -875,6 +875,66 @@ public void testExtractorTimeRangeMatch() throws Exception {
     }
   }
 
+  @Test
+  public void testSourceTimeRangeRespectsHistoryDisable() throws Exception {
+    final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+    final String receiverIp = receiverDataNode.getIp();
+    final int receiverPort = receiverDataNode.getPort();
+
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "insert into root.db.history (time, at1) values (2000, 2), (3000, 3)", "flush"),
+        null);
+
+    final Map<String, String> sourceAttributes = new HashMap<>();
+    final Map<String, String> sinkAttributes = new HashMap<>();
+
+    sourceAttributes.put("source.inclusion", "data");
+    sourceAttributes.put("source.start-time", "2000");
+    sourceAttributes.put("source.history.enable", "false");
+    sourceAttributes.put("source.realtime.mode", "stream");
+    sourceAttributes.put("user", "root");
+
+    sinkAttributes.put("sink", "iotdb-thrift-sink");
+    sinkAttributes.put("sink.batch.enable", "false");
+    sinkAttributes.put("sink.ip", receiverIp);
+    sinkAttributes.put("sink.port", Integer.toString(receiverPort));
+
+    try (final SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+      final TSStatus status =
+          client.createPipe(
+              new TCreatePipeReq("p1", sinkAttributes).setExtractorAttributes(sourceAttributes));
+      Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv,
+          "show timeseries root.db.history.**",
+          "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
+          Collections.emptySet());
+
+      TestUtils.executeNonQueries(
+          senderEnv,
+          Collections.singletonList(
+              "insert into root.db.realtime (time, at1)"
+                  + " values (1000, 1), (2000, 2), (3000, 3)"),
+          null);
+
+      TestUtils.assertDataEventuallyOnEnv(
+          receiverEnv,
+          "select count(at1) from root.db.realtime",
+          "count(root.db.realtime.at1),",
+          Collections.singleton("2,"));
+      TestUtils.assertDataAlwaysOnEnv(
+          receiverEnv,
+          "show timeseries root.db.history.**",
+          "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
+          Collections.emptySet());
+    }
+  }
+
   @Test
   public void testSourceStartTimeAndEndTimeWorkingWithOrWithoutPattern() throws Exception {
     final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java

@@ -172,7 +172,7 @@ public void validate(final PipeParameterValidator validator) throws Exception {
           EXTRACTOR_REALTIME_MODE_BATCH_MODE_VALUE);
     }
 
-    // Validate source.start-time and source.end-time
+    // A global time range takes precedence over a history-specific time range.
     if (validator
             .getParameters()
             .hasAnyAttributes(
@@ -183,20 +183,16 @@ public void validate(final PipeParameterValidator validator) throws Exception {
         && validator
             .getParameters()
             .hasAnyAttributes(
-                EXTRACTOR_HISTORY_ENABLE_KEY,
-                SOURCE_HISTORY_ENABLE_KEY,
                 SOURCE_HISTORY_START_TIME_KEY,
                 EXTRACTOR_HISTORY_START_TIME_KEY,
                 SOURCE_HISTORY_END_TIME_KEY,
                 EXTRACTOR_HISTORY_END_TIME_KEY)) {
       LOGGER.warn(
-          "When {}, {}, {} or {} is specified, specifying {}, {}, {}, {}, {} and {} is invalid.",
+          "When {}, {}, {} or {} is specified, specifying {}, {}, {} or {} is invalid.",
           SOURCE_START_TIME_KEY,
           EXTRACTOR_START_TIME_KEY,
           SOURCE_END_TIME_KEY,
           EXTRACTOR_END_TIME_KEY,
-          SOURCE_HISTORY_ENABLE_KEY,
-          EXTRACTOR_HISTORY_ENABLE_KEY,
           SOURCE_HISTORY_START_TIME_KEY,
           EXTRACTOR_HISTORY_START_TIME_KEY,
           SOURCE_HISTORY_END_TIME_KEY,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java

@@ -165,13 +165,23 @@ public void validate(final PipeParameterValidator validator) {
       }
     }
 
+    // Historical data extraction is enabled in the following cases:
+    // 1. System restarts the pipe. If the pipe is restarted but historical data extraction is not
+    // enabled, the pipe will lose some historical data.
+    // 2. Historical extraction is enabled by the user or by default.
+    isHistoricalSourceEnabled =
+        parameters.getBooleanOrDefault(
+                SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
+                SystemConstant.RESTART_OR_NEWLY_ADDED_DEFAULT_VALUE)
+            || parameters.getBooleanOrDefault(
+                Arrays.asList(EXTRACTOR_HISTORY_ENABLE_KEY, SOURCE_HISTORY_ENABLE_KEY),
+                EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE);
+
     if (parameters.hasAnyAttributes(
         SOURCE_START_TIME_KEY,
         EXTRACTOR_START_TIME_KEY,
         SOURCE_END_TIME_KEY,
         EXTRACTOR_END_TIME_KEY)) {
-      isHistoricalSourceEnabled = true;
-
       try {
         historicalDataExtractionStartTime =
             parameters.hasAnyAttributes(SOURCE_START_TIME_KEY, EXTRACTOR_START_TIME_KEY)
@@ -205,19 +215,6 @@ public void validate(final PipeParameterValidator validator) {
       return;
     }
 
-    // Historical data extraction is enabled in the following cases:
-    // 1. System restarts the pipe. If the pipe is restarted but historical data extraction is not
-    // enabled, the pipe will lose some historical data.
-    // 2. User may set the EXTRACTOR_HISTORY_START_TIME and EXTRACTOR_HISTORY_END_TIME without
-    // enabling the historical data extraction, which may affect the realtime data extraction.
-    isHistoricalSourceEnabled =
-        parameters.getBooleanOrDefault(
-                SystemConstant.RESTART_OR_NEWLY_ADDED_KEY,
-                SystemConstant.RESTART_OR_NEWLY_ADDED_DEFAULT_VALUE)
-            || parameters.getBooleanOrDefault(
-                Arrays.asList(EXTRACTOR_HISTORY_ENABLE_KEY, SOURCE_HISTORY_ENABLE_KEY),
-                EXTRACTOR_HISTORY_ENABLE_DEFAULT_VALUE);
-
     try {
       historicalDataExtractionStartTime =
           parameters.hasAnyAttributes(

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java

@@ -24,10 +24,14 @@
 import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
 import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
 import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixPipePattern;
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 
 import com.google.common.collect.ImmutableMap;
 import org.apache.tsfile.file.metadata.PlainDeviceID;
@@ -38,9 +42,46 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.Map;
 
 public class PipeHistoricalDataRegionTsFileSourceTest {
 
+  @Test
+  public void testGlobalTimeRangeRespectsHistoryEnable() throws Exception {
+    final Map<String, String> attributes = new HashMap<>();
+    attributes.put(PipeSourceConstant.SOURCE_START_TIME_KEY, "1000");
+    attributes.put(PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY, Boolean.FALSE.toString());
+
+    final PipeHistoricalDataRegionTsFileSource realtimeOnlySource =
+        new PipeHistoricalDataRegionTsFileSource();
+    realtimeOnlySource.validate(
+        new PipeParameterValidator(new PipeParameters(new HashMap<>(attributes))));
+
+    Assert.assertFalse((Boolean) getPrivateField(realtimeOnlySource, "isHistoricalSourceEnabled"));
+    Assert.assertEquals(
+        1000L,
+        ((Long) getPrivateField(realtimeOnlySource, "historicalDataExtractionStartTime"))
+            .longValue());
+
+    final PipeHistoricalDataRegionTsFileSource defaultSource =
+        new PipeHistoricalDataRegionTsFileSource();
+    attributes.remove(PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY);
+    defaultSource.validate(
+        new PipeParameterValidator(new PipeParameters(new HashMap<>(attributes))));
+
+    Assert.assertTrue((Boolean) getPrivateField(defaultSource, "isHistoricalSourceEnabled"));
+
+    final PipeHistoricalDataRegionTsFileSource restartedSource =
+        new PipeHistoricalDataRegionTsFileSource();
+    attributes.put(PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY, Boolean.FALSE.toString());
+    attributes.put(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY, Boolean.TRUE.toString());
+    restartedSource.validate(
+        new PipeParameterValidator(new PipeParameters(new HashMap<>(attributes))));
+
+    Assert.assertTrue((Boolean) getPrivateField(restartedSource, "isHistoricalSourceEnabled"));
+  }
+
   @Test
   public void testMayTsFileContainUnprocessedDataUsesEqualOrAfterCoverage() throws Exception {
     final File tempDir = Files.createTempDirectory("pipeHistoricalProgressCoverage").toFile();
@@ -166,4 +207,12 @@ private static void setPrivateField(
     field.setAccessible(true);
     field.set(source, value);
   }
+
+  private static Object getPrivateField(
+      final PipeHistoricalDataRegionTsFileSource source, final String fieldName)
+      throws ReflectiveOperationException {
+    final Field field = PipeHistoricalDataRegionTsFileSource.class.getDeclaredField(fieldName);
+    field.setAccessible(true);
+    return field.get(source);
+  }
 }