From ed1369f5ba98e00ee4cff2dd44baefcabc0c507d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 30 Jul 2026 17:27:22 +0800 Subject: [PATCH] Pipe: Respect history enable with source time range --- .../auto/basic/IoTDBPipeSourceIT.java | 60 +++++++++++++++++++ .../iotdb/db/i18n/DataNodePipeMessages.java | 4 +- .../iotdb/db/i18n/DataNodePipeMessages.java | 4 +- .../dataregion/IoTDBDataRegionSource.java | 8 +-- ...icalDataRegionTsFileAndDeletionSource.java | 27 ++++----- ...DataRegionTsFileAndDeletionSourceTest.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 fce4361163fec..6f1405782823c 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 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 sourceAttributes = new HashMap<>(); + final Map 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 b75097a50cda7..d9a954b57e7a4 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 @@ -651,8 +651,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 f03d649ac81d8..13b262a0710ef 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 @@ -616,8 +616,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 3e9c8ea4f7e9d..6adcb1addb522 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 @@ private void validatePattern(final TreePattern treePattern) { 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 b5ab9e0c65e06..f4f4eda2e0e8f 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 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) @@ -284,19 +294,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( 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 9d5c895ce83ee..0dd6bae7ddfc9 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.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 void testHistoricalTsFileQueryPriorityOrderDefaultsToTrue() throws Except (Boolean) getPrivateField(source, "shouldOrderHistoricalTsFileByQueryPriority")); } + @Test + public void testGlobalTimeRangeRespectsHistoryEnable() throws Exception { + final Map 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 =