This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch pipe-sloppy-time-range in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 28451d4cd948846b7279c293015c8d145ccea851 Author: Steve Yurong Su <[email protected]> AuthorDate: Wed Oct 25 16:04:00 2023 +0800 Pipe: support parameter extractor.history.sloppy-time-range to improve historical data processing efficiency --- .../iotdb/db/pipe/config/constant/PipeExtractorConstant.java | 2 ++ .../historical/PipeHistoricalDataRegionTsFileExtractor.java | 7 ++++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeExtractorConstant.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeExtractorConstant.java index 6a7c095c148..b4be2ed34b9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeExtractorConstant.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/config/constant/PipeExtractorConstant.java @@ -33,6 +33,8 @@ public class PipeExtractorConstant { public static final String EXTRACTOR_HISTORY_ENABLE_KEY = "extractor.history.enable"; public static final String EXTRACTOR_HISTORY_START_TIME = "extractor.history.start-time"; public static final String EXTRACTOR_HISTORY_END_TIME = "extractor.history.end-time"; + public static final String EXTRACTOR_HISTORY_SLOPPY_TIME_RANGE = + "extractor.history.sloppy-time-range"; public static final String EXTRACTOR_REALTIME_ENABLE = "extractor.realtime.enable"; public static final String EXTRACTOR_REALTIME_MODE = "extractor.realtime.mode"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java index b250436622e..1a704b44722 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/historical/PipeHistoricalDataRegionTsFileExtractor.java @@ -50,6 +50,7 @@ import java.util.stream.Collectors; import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_ENABLE_KEY; import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_END_TIME; +import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_SLOPPY_TIME_RANGE; import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_HISTORY_START_TIME; import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_PATTERN_DEFAULT_VALUE; import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_PATTERN_KEY; @@ -74,6 +75,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa private long historicalDataExtractionTimeLowerBound; // Arrival time + private boolean sloppyTimeRange; // true to disable time range filter after extraction + private Queue<TsFileResource> pendingQueue; @Override @@ -151,6 +154,8 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa } } } + + sloppyTimeRange = parameters.getBooleanOrDefault(EXTRACTOR_HISTORY_SLOPPY_TIME_RANGE, false); } private void flushDataRegionAllTsFiles() { @@ -289,7 +294,7 @@ public class PipeHistoricalDataRegionTsFileExtractor implements PipeHistoricalDa pattern, historicalDataExtractionStartTime, historicalDataExtractionEndTime, - !isTsFileResourceCoveredByTimeRange(resource)); + !sloppyTimeRange && !isTsFileResourceCoveredByTimeRange(resource)); event.increaseReferenceCount(PipeHistoricalDataRegionTsFileExtractor.class.getName()); try { PipeResourceManager.tsfile().unpinTsFileResource(resource);
