This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 154644107a8 Pipe: Respect history enable with source time range
(#18362)
154644107a8 is described below
commit 154644107a82ba69431725390b8346149263bd0e
Author: Caideyipi <[email protected]>
AuthorDate: Fri Jul 31 10:57:56 2026 +0800
Pipe: Respect history enable with source time range (#18362)
---
.../treemodel/auto/basic/IoTDBPipeSourceIT.java | 60 ++++++++++++++++++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 +-
.../source/dataregion/IoTDBDataRegionSource.java | 8 +--
...istoricalDataRegionTsFileAndDeletionSource.java | 27 +++++-----
...ricalDataRegionTsFileAndDeletionSourceTest.java | 36 +++++++++++++
6 files changed, 114 insertions(+), 25 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
index fce4361163f..6f140578282 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSourceIT.java
@@ -895,6 +895,66 @@ public class IoTDBPipeSourceIT extends
AbstractPipeDualTreeModelAutoIT {
}
}
+ @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/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index abda0e8031f..8e61bcda388 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -653,8 +653,8 @@ public final class DataNodePipeMessages {
"When '{}' ('{}') is set to false, specifying {} and {} is invalid.";
public static final String WHEN_IS_SET_TO_TRUE_SPECIFYING_AND =
"When '{}' ('{}', '{}', '{}') is set to true, specifying {} and {} is
invalid.";
- public static final String WHEN_OR_IS_SPECIFIED_SPECIFYING_AND_IS =
- "When {}, {}, {} or {} is specified, specifying {}, {}, {}, {}, {} and
{} is invalid.";
+ public static final String WHEN_OR_IS_SPECIFIED_SPECIFYING_OR_IS_INVALID =
+ "When {}, {}, {} or {} is specified, specifying {}, {}, {} or {} is
invalid.";
// ===================== SINK =====================
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 11604aef115..2389f97c8ec 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -618,8 +618,8 @@ public final class DataNodePipeMessages {
"当 '{}'（'{}'）设置为 false 时，指定 {} 和 {} 无效。";
public static final String WHEN_IS_SET_TO_TRUE_SPECIFYING_AND =
"当 '{}'（'{}'、'{}'、'{}'）设置为 true 时，指定 {} 和 {} 无效。";
- public static final String WHEN_OR_IS_SPECIFIED_SPECIFYING_AND_IS =
- "当指定 {}、{}、{} 或 {} 时，指定 {}、{}、{}、{}、{} 和 {} 无效。";
+ public static final String WHEN_OR_IS_SPECIFIED_SPECIFYING_OR_IS_INVALID =
+ "当指定 {}、{}、{} 或 {} 时，指定 {}、{}、{} 或 {} 无效。";
// ===================== SINK =====================
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 3e9c8ea4f7e..6adcb1addb5 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
@@ -257,27 +257,23 @@ public class IoTDBDataRegionSource extends IoTDBSource {
private void checkInvalidParameters(final PipeParameterValidator validator) {
final PipeParameters parameters = validator.getParameters();
- // Enable history and realtime if specifying start-time or end-time
+ // A global time range takes precedence over a history-specific time range.
if (parameters.hasAnyAttributes(
SOURCE_START_TIME_KEY,
EXTRACTOR_START_TIME_KEY,
SOURCE_END_TIME_KEY,
EXTRACTOR_END_TIME_KEY)
&& parameters.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(
- DataNodePipeMessages.WHEN_OR_IS_SPECIFIED_SPECIFYING_AND_IS,
+ DataNodePipeMessages.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/PipeHistoricalDataRegionTsFileAndDeletionSource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
index bfd705f7419..cf913176df8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java
@@ -243,13 +243,23 @@ public class
PipeHistoricalDataRegionTsFileAndDeletionSource
}
}
+ // 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)
@@ -284,19 +294,6 @@ public class
PipeHistoricalDataRegionTsFileAndDeletionSource
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/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java
index 9d5c895ce83..0dd6bae7ddf 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSourceTest.java
@@ -28,6 +28,7 @@ import
org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
import
org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
+import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import
org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeConfiguration;
import
org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
import org.apache.iotdb.commons.pipe.datastructure.resource.PersistentResource;
@@ -193,6 +194,41 @@ public class
PipeHistoricalDataRegionTsFileAndDeletionSourceTest {
(Boolean) getPrivateField(source,
"shouldOrderHistoricalTsFileByQueryPriority"));
}
+ @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 PipeHistoricalDataRegionTsFileAndDeletionSource realtimeOnlySource =
+ new PipeHistoricalDataRegionTsFileAndDeletionSource();
+ realtimeOnlySource.validate(
+ new PipeParameterValidator(new PipeParameters(new
HashMap<>(attributes))));
+
+ Assert.assertFalse((Boolean) getPrivateField(realtimeOnlySource,
"isHistoricalSourceEnabled"));
+ Assert.assertEquals(
+ 1000L,
+ ((Long) getPrivateField(realtimeOnlySource,
"historicalDataExtractionStartTime"))
+ .longValue());
+
+ final PipeHistoricalDataRegionTsFileAndDeletionSource defaultSource =
+ new PipeHistoricalDataRegionTsFileAndDeletionSource();
+ attributes.remove(PipeSourceConstant.SOURCE_HISTORY_ENABLE_KEY);
+ defaultSource.validate(
+ new PipeParameterValidator(new PipeParameters(new
HashMap<>(attributes))));
+
+ Assert.assertTrue((Boolean) getPrivateField(defaultSource,
"isHistoricalSourceEnabled"));
+
+ final PipeHistoricalDataRegionTsFileAndDeletionSource restartedSource =
+ new PipeHistoricalDataRegionTsFileAndDeletionSource();
+ 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 testHistoricalTsFileQueryPriorityOrderMatchesQueryCoverage()
throws Exception {
final PipeHistoricalDataRegionTsFileAndDeletionSource source =
