From 9607cbab0b7ad72df9a4e68c1908171a6f050ce2 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:57:56 +0800 Subject: [PATCH] Pipe: Respect history enable with source time range (#18362) --- .../pipe/it/autocreate/IoTDBPipeSourceIT.java | 60 +++++++++++++++++++ .../dataregion/IoTDBDataRegionSource.java | 8 +-- .../PipeHistoricalDataRegionTsFileSource.java | 27 ++++----- ...eHistoricalDataRegionTsFileSourceTest.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 c592aca4e6afe..b0428ce674dcb 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 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/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 6df5e10d0a254..9ec1e2c34d2b8 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 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, 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 00fd08dc2b7b4..d0b3abfb70703 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 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( 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 a00fc2c19dbf8..003d8f09c5408 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.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 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); + } }