You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/07/25 09:11:19 UTC
[iotdb] branch rel/1.2 updated: [IOTDB-6081] Pipe: use HybridExtractor instead of LogExtractor when realtime mode is set to log to avoid OOM under heavy insertion load (#10620) (#10669)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new 2c51815ac63 [IOTDB-6081] Pipe: use HybridExtractor instead of LogExtractor when realtime mode is set to log to avoid OOM under heavy insertion load (#10620) (#10669)
2c51815ac63 is described below
commit 2c51815ac636c06b37a44a6b2a479fd8ecd40f9f
Author: 马子坤 <55...@users.noreply.github.com>
AuthorDate: Tue Jul 25 17:11:11 2023 +0800
[IOTDB-6081] Pipe: use HybridExtractor instead of LogExtractor when realtime mode is set to log to avoid OOM under heavy insertion load (#10620) (#10669)
* When user specifies 'extractor.realtime.mode'='log', we use hybrid extractors which can automatically degrade to file mode when load is too heavy.
* Add a parameter: 'extractor.realtime.mode'='forced-log', the pipe will always use log mode and will never degrade to file mode.
---------
Co-authored-by: Steve Yurong Su <ro...@apache.org>
(cherry picked from commit 01378cc91b0a4349f8a714754f82054ec9b425fb)
---
.../org/apache/iotdb/pipe/api/PipeConnector.java | 7 ++++++-
.../org/apache/iotdb/pipe/api/PipeProcessor.java | 9 +++++++--
.../config/constant/PipeExtractorConstant.java | 1 +
.../pipe/extractor/IoTDBDataRegionExtractor.java | 23 ++++++++++++++--------
.../extractor/IoTDBDataRegionExtractorTest.java | 2 +-
5 files changed, 30 insertions(+), 12 deletions(-)
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
index 5d14da1e3aa..26ac2d5f031 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeConnector.java
@@ -127,7 +127,12 @@ public interface PipeConnector extends PipePlugin {
* @throws PipeConnectionException if the connection is broken
* @throws Exception the user can throw errors if necessary
*/
- void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception;
+ default void transfer(TsFileInsertionEvent tsFileInsertionEvent) throws Exception {
+ for (final TabletInsertionEvent tabletInsertionEvent :
+ tsFileInsertionEvent.toTabletInsertionEvents()) {
+ transfer(tabletInsertionEvent);
+ }
+ }
/**
* This method is used to transfer the Event.
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeProcessor.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeProcessor.java
index f752aafef3a..b8388992a17 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeProcessor.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeProcessor.java
@@ -103,8 +103,13 @@ public interface PipeProcessor extends PipePlugin {
* @param eventCollector used to collect result events after processing
* @throws Exception the user can throw errors if necessary
*/
- void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector)
- throws Exception;
+ default void process(TsFileInsertionEvent tsFileInsertionEvent, EventCollector eventCollector)
+ throws Exception {
+ for (final TabletInsertionEvent tabletInsertionEvent :
+ tsFileInsertionEvent.toTabletInsertionEvents()) {
+ process(tabletInsertionEvent, eventCollector);
+ }
+ }
/**
* This method is called to process the Event.
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 6b003860352..b92b502ae19 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
@@ -35,6 +35,7 @@ public class PipeExtractorConstant {
public static final String EXTRACTOR_REALTIME_MODE_HYBRID = "hybrid";
public static final String EXTRACTOR_REALTIME_MODE_FILE = "file";
public static final String EXTRACTOR_REALTIME_MODE_LOG = "log";
+ public static final String EXTRACTOR_REALTIME_MODE_FORCED_LOG = "forced-log";
private PipeExtractorConstant() {
throw new IllegalStateException("Utility class");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java
index 580b31cde47..35749c31a83 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractor.java
@@ -46,6 +46,7 @@ import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXT
import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_ENABLE;
import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE;
import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_FILE;
+import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_FORCED_LOG;
import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_HYBRID;
import static org.apache.iotdb.db.pipe.config.constant.PipeExtractorConstant.EXTRACTOR_REALTIME_MODE_LOG;
@@ -85,9 +86,10 @@ public class IoTDBDataRegionExtractor implements PipeExtractor {
validator.validateAttributeValueRange(
EXTRACTOR_REALTIME_MODE,
true,
- EXTRACTOR_REALTIME_MODE_HYBRID,
EXTRACTOR_REALTIME_MODE_FILE,
- EXTRACTOR_REALTIME_MODE_LOG);
+ EXTRACTOR_REALTIME_MODE_HYBRID,
+ EXTRACTOR_REALTIME_MODE_LOG,
+ EXTRACTOR_REALTIME_MODE_FORCED_LOG);
}
constructHistoricalExtractor();
@@ -106,12 +108,14 @@ public class IoTDBDataRegionExtractor implements PipeExtractor {
// Enable realtime extractor by default
if (!parameters.getBooleanOrDefault(EXTRACTOR_REALTIME_ENABLE, true)) {
realtimeExtractor = new PipeRealtimeDataRegionFakeExtractor();
+ LOGGER.info("'{}' is set to false, use fake realtime extractor.", EXTRACTOR_REALTIME_ENABLE);
return;
}
// Use hybrid mode by default
if (!parameters.hasAttribute(EXTRACTOR_REALTIME_MODE)) {
realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
+ LOGGER.info("'{}' is not set, use hybrid mode by default.", EXTRACTOR_REALTIME_MODE);
return;
}
@@ -119,17 +123,20 @@ public class IoTDBDataRegionExtractor implements PipeExtractor {
case EXTRACTOR_REALTIME_MODE_FILE:
realtimeExtractor = new PipeRealtimeDataRegionTsFileExtractor();
break;
- case EXTRACTOR_REALTIME_MODE_LOG:
- realtimeExtractor = new PipeRealtimeDataRegionLogExtractor();
- break;
case EXTRACTOR_REALTIME_MODE_HYBRID:
+ case EXTRACTOR_REALTIME_MODE_LOG:
realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
break;
+ case EXTRACTOR_REALTIME_MODE_FORCED_LOG:
+ realtimeExtractor = new PipeRealtimeDataRegionLogExtractor();
+ break;
default:
realtimeExtractor = new PipeRealtimeDataRegionHybridExtractor();
- LOGGER.warn(
- "Unsupported extractor realtime mode: {}, create a hybrid extractor.",
- parameters.getString(EXTRACTOR_REALTIME_MODE));
+ if (LOGGER.isWarnEnabled()) {
+ LOGGER.warn(
+ "Unsupported extractor realtime mode: {}, create a hybrid extractor.",
+ parameters.getString(EXTRACTOR_REALTIME_MODE));
+ }
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java
index 5dfcf5f5568..3d4bbbbb9b3 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/extractor/IoTDBDataRegionExtractorTest.java
@@ -30,7 +30,7 @@ import java.util.HashMap;
public class IoTDBDataRegionExtractorTest {
@Test
- public void testIoTDBDataRegionExtractorTest() {
+ public void testIoTDBDataRegionExtractor() {
IoTDBDataRegionExtractor extractor = new IoTDBDataRegionExtractor();
try {
extractor.validate(