This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 5b0894730cb Pipe: Respect history enable with source time range
(#18362) (#18385)
5b0894730cb is described below
commit 5b0894730cb09a9a2b5b912ebc03ddf35373e8b2
Author: Caideyipi <[email protected]>
AuthorDate: Tue Aug 4 12:18:04 2026 +0800
Pipe: Respect history enable with source time range (#18362) (#18385)
---
.../pipe/it/autocreate/IoTDBPipeSourceIT.java | 60 ++++++++++++++++++++++
.../source/dataregion/IoTDBDataRegionSource.java | 8 +--
.../PipeHistoricalDataRegionTsFileSource.java | 27 +++++-----
.../PipeHistoricalDataRegionTsFileSourceTest.java | 49 ++++++++++++++++++
4 files changed, 123 insertions(+), 21 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSourceIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSourceIT.java
index c592aca4e6a..b0428ce674d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSourceIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSourceIT.java
@@ -875,6 +875,66 @@ public class IoTDBPipeSourceIT extends
AbstractPipeDualAutoIT {
}
}
+ @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);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
index 6df5e10d0a2..9ec1e2c34d2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
@@ -172,7 +172,7 @@ public class IoTDBDataRegionSource extends IoTDBSource {
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 class IoTDBDataRegionSource extends IoTDBSource {
&& 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,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
index 6ebdebb6efd..25e65ed8be9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSource.java
@@ -165,13 +165,23 @@ public class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
}
}
+ // 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 class PipeHistoricalDataRegionTsFileSource
implements PipeHistoricalDataR
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(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java
index a00fc2c19db..003d8f09c54 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileSourceTest.java
+++
b/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.HybridProgressIndex;
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.io.File;
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 @@ public class PipeHistoricalDataRegionTsFileSourceTest {
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);
+ }
}
