Skip to content

Commit caca491

Browse files
CaideyipiCaideyipi
authored
Pipe: Improved the total performance by disable some useless logic (#16916)
1 parent 19885ea commit caca491

File tree

10 files changed

+30
-684
lines changed

10 files changed

+30
-684
lines changed

‎.github/workflows/pipe-it-2cluster.yml‎

Lines changed: 0 additions & 539 deletions
This file was deleted.

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java‎

Lines changed: 1 addition & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -332,11 +332,6 @@ public void testAsyncConnectorUseNodeUrls() throws Exception {
332332
doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName());
333333
}
334334

335-
@Test
336-
public void testAirGapConnectorUseNodeUrls() throws Exception {
337-
doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName());
338-
}
339-
340335
private void doTestUseNodeUrls(String connectorName) throws Exception {
341336
senderEnv
342337
.getConfig()
@@ -371,16 +366,7 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {
371366

372367
final StringBuilder nodeUrlsBuilder = new StringBuilder();
373368
for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) {
374-
if (connectorName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) {
375-
// Use default port for convenience
376-
nodeUrlsBuilder
377-
.append(wrapper.getIp())
378-
.append(":")
379-
.append(wrapper.getPipeAirGapReceiverPort())
380-
.append(",");
381-
} else {
382-
nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
383-
}
369+
nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
384370
}
385371

386372
try (final SyncConfigNodeIServiceClient client =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/plugin/PipeConfigRegionSinkConstructor.java‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
2323
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink;
2424
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor;
25-
import org.apache.iotdb.confignode.manager.pipe.sink.protocol.IoTDBConfigRegionAirGapSink;
2625
import org.apache.iotdb.confignode.manager.pipe.sink.protocol.IoTDBConfigRegionSink;
2726
import org.apache.iotdb.pipe.api.PipeConnector;
2827

@@ -41,9 +40,6 @@ protected void initConstructors() {
4140
pluginConstructors.put(
4241
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
4342
IoTDBConfigRegionSink::new);
44-
pluginConstructors.put(
45-
BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(),
46-
IoTDBConfigRegionAirGapSink::new);
4743
pluginConstructors.put(
4844
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new);
4945

@@ -55,8 +51,6 @@ protected void initConstructors() {
5551
BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(), IoTDBConfigRegionSink::new);
5652
pluginConstructors.put(
5753
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(), IoTDBConfigRegionSink::new);
58-
pluginConstructors.put(
59-
BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBConfigRegionAirGapSink::new);
6054
pluginConstructors.put(
6155
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new);
6256
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionProcessorConstructor.java‎

Lines changed: 0 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -21,17 +21,9 @@
2121

2222
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
2323
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.donothing.DoNothingProcessor;
24-
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.processor.throwing.ThrowingExceptionProcessor;
2524
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeProcessorConstructor;
2625
import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper;
27-
import org.apache.iotdb.db.pipe.processor.aggregate.AggregateProcessor;
28-
import org.apache.iotdb.db.pipe.processor.aggregate.operator.processor.StandardStatisticsOperatorProcessor;
29-
import org.apache.iotdb.db.pipe.processor.aggregate.window.processor.TumblingWindowingProcessor;
30-
import org.apache.iotdb.db.pipe.processor.downsampling.changing.ChangingValueSamplingProcessor;
31-
import org.apache.iotdb.db.pipe.processor.downsampling.sdt.SwingingDoorTrendingSamplingProcessor;
32-
import org.apache.iotdb.db.pipe.processor.downsampling.tumbling.TumblingTimeSamplingProcessor;
3326
import org.apache.iotdb.db.pipe.processor.pipeconsensus.PipeConsensusProcessor;
34-
import org.apache.iotdb.db.pipe.processor.twostage.plugin.TwoStageCountProcessor;
3527

3628
class PipeDataRegionProcessorConstructor extends PipeProcessorConstructor {
3729

@@ -43,28 +35,6 @@ class PipeDataRegionProcessorConstructor extends PipeProcessorConstructor {
4335
protected void initConstructors() {
4436
pluginConstructors.put(
4537
BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName(), DoNothingProcessor::new);
46-
pluginConstructors.put(
47-
BuiltinPipePlugin.TUMBLING_TIME_SAMPLING_PROCESSOR.getPipePluginName(),
48-
TumblingTimeSamplingProcessor::new);
49-
pluginConstructors.put(
50-
BuiltinPipePlugin.SDT_SAMPLING_PROCESSOR.getPipePluginName(),
51-
SwingingDoorTrendingSamplingProcessor::new);
52-
pluginConstructors.put(
53-
BuiltinPipePlugin.CHANGING_VALUE_SAMPLING_PROCESSOR.getPipePluginName(),
54-
ChangingValueSamplingProcessor::new);
55-
pluginConstructors.put(
56-
BuiltinPipePlugin.THROWING_EXCEPTION_PROCESSOR.getPipePluginName(),
57-
ThrowingExceptionProcessor::new);
58-
pluginConstructors.put(
59-
BuiltinPipePlugin.AGGREGATE_PROCESSOR.getPipePluginName(), AggregateProcessor::new);
60-
pluginConstructors.put(
61-
BuiltinPipePlugin.STANDARD_STATISTICS_PROCESSOR.getPipePluginName(),
62-
StandardStatisticsOperatorProcessor::new);
63-
pluginConstructors.put(
64-
BuiltinPipePlugin.TUMBLING_WINDOWING_PROCESSOR.getPipePluginName(),
65-
TumblingWindowingProcessor::new);
66-
pluginConstructors.put(
67-
BuiltinPipePlugin.COUNT_POINT_PROCESSOR.getPipePluginName(), TwoStageCountProcessor::new);
6838
pluginConstructors.put(
6939
BuiltinPipePlugin.PIPE_CONSENSUS_PROCESSOR.getPipePluginName(),
7040
PipeConsensusProcessor::new);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/dataregion/PipeDataRegionSinkConstructor.java‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,7 @@
2323
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink;
2424
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor;
2525
import org.apache.iotdb.commons.pipe.agent.plugin.meta.DataNodePipePluginMetaKeeper;
26-
import org.apache.iotdb.db.pipe.sink.protocol.airgap.IoTDBDataRegionAirGapSink;
2726
import org.apache.iotdb.db.pipe.sink.protocol.legacy.IoTDBLegacyPipeSink;
28-
import org.apache.iotdb.db.pipe.sink.protocol.opcda.OpcDaSink;
29-
import org.apache.iotdb.db.pipe.sink.protocol.opcua.OpcUaSink;
3027
import org.apache.iotdb.db.pipe.sink.protocol.pipeconsensus.PipeConsensusAsyncSink;
3128
import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
3229
import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
@@ -59,13 +56,8 @@ protected void initConstructors() {
5956
pluginConstructors.put(
6057
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_CONNECTOR.getPipePluginName(),
6158
IoTDBLegacyPipeSink::new);
62-
pluginConstructors.put(
63-
BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(),
64-
IoTDBDataRegionAirGapSink::new);
6559
pluginConstructors.put(
6660
BuiltinPipePlugin.WEBSOCKET_CONNECTOR.getPipePluginName(), WebSocketSink::new);
67-
pluginConstructors.put(BuiltinPipePlugin.OPC_UA_CONNECTOR.getPipePluginName(), OpcUaSink::new);
68-
pluginConstructors.put(BuiltinPipePlugin.OPC_DA_CONNECTOR.getPipePluginName(), OpcDaSink::new);
6961
pluginConstructors.put(
7062
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new);
7163
pluginConstructors.put(
@@ -82,12 +74,8 @@ protected void initConstructors() {
8274
IoTDBDataRegionAsyncSink::new);
8375
pluginConstructors.put(
8476
BuiltinPipePlugin.IOTDB_LEGACY_PIPE_SINK.getPipePluginName(), IoTDBLegacyPipeSink::new);
85-
pluginConstructors.put(
86-
BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBDataRegionAirGapSink::new);
8777
pluginConstructors.put(
8878
BuiltinPipePlugin.WEBSOCKET_SINK.getPipePluginName(), WebSocketSink::new);
89-
pluginConstructors.put(BuiltinPipePlugin.OPC_UA_SINK.getPipePluginName(), OpcUaSink::new);
90-
pluginConstructors.put(BuiltinPipePlugin.OPC_DA_SINK.getPipePluginName(), OpcDaSink::new);
9179
pluginConstructors.put(
9280
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new);
9381
pluginConstructors.put(

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/plugin/schemaregion/PipeSchemaRegionSinkConstructor.java‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
2323
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.sink.donothing.DoNothingSink;
2424
import org.apache.iotdb.commons.pipe.agent.plugin.constructor.PipeSinkConstructor;
25-
import org.apache.iotdb.db.pipe.sink.protocol.airgap.IoTDBSchemaRegionAirGapSink;
2625
import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBSchemaRegionSink;
2726
import org.apache.iotdb.pipe.api.PipeConnector;
2827

@@ -41,9 +40,6 @@ protected void initConstructors() {
4140
pluginConstructors.put(
4241
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
4342
IoTDBSchemaRegionSink::new);
44-
pluginConstructors.put(
45-
BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName(),
46-
IoTDBSchemaRegionAirGapSink::new);
4743
pluginConstructors.put(
4844
BuiltinPipePlugin.DO_NOTHING_CONNECTOR.getPipePluginName(), DoNothingSink::new);
4945

@@ -55,8 +51,6 @@ protected void initConstructors() {
5551
BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(), IoTDBSchemaRegionSink::new);
5652
pluginConstructors.put(
5753
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName(), IoTDBSchemaRegionSink::new);
58-
pluginConstructors.put(
59-
BuiltinPipePlugin.IOTDB_AIR_GAP_SINK.getPipePluginName(), IoTDBSchemaRegionAirGapSink::new);
6054
pluginConstructors.put(
6155
BuiltinPipePlugin.DO_NOTHING_SINK.getPipePluginName(), DoNothingSink::new);
6256
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java‎

Lines changed: 25 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121

2222
import org.apache.iotdb.commons.consensus.DataRegionId;
2323
import org.apache.iotdb.commons.pipe.agent.task.PipeTaskAgent;
24+
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
25+
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
2426
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePatternOperations;
2527
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
2628
import org.apache.iotdb.commons.pipe.source.IoTDBSource;
@@ -45,6 +47,7 @@
4547
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
4648
import org.apache.iotdb.pipe.api.event.dml.insertion.TsFileInsertionEvent;
4749
import org.apache.iotdb.pipe.api.exception.PipeException;
50+
import org.apache.iotdb.pipe.api.exception.PipeParameterNotValidException;
4851

4952
import org.apache.tsfile.utils.Pair;
5053
import org.slf4j.Logger;
@@ -100,6 +103,21 @@ public class IoTDBDataRegionSource extends IoTDBSource {
100103
public void validate(final PipeParameterValidator validator) throws Exception {
101104
super.validate(validator);
102105

106+
final boolean forwardingPipeRequests =
107+
validator
108+
.getParameters()
109+
.getBooleanOrDefault(
110+
Arrays.asList(
111+
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
112+
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
113+
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
114+
if (!forwardingPipeRequests) {
115+
throw new PipeParameterNotValidException(
116+
String.format(
117+
"The parameter %s cannot be set to false.",
118+
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY));
119+
}
120+
103121
final Pair<Boolean, Boolean> insertionDeletionListeningOptionPair =
104122
DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(
105123
validator.getParameters());
@@ -163,26 +181,6 @@ public void validate(final PipeParameterValidator validator) throws Exception {
163181
Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY),
164182
EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE));
165183

166-
// Validate source.realtime.mode
167-
if (validator
168-
.getParameters()
169-
.getBooleanOrDefault(
170-
Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY),
171-
EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE)
172-
|| validator.getParameters().hasAnyAttributes(SOURCE_START_TIME_KEY, SOURCE_END_TIME_KEY)) {
173-
validator.validateAttributeValueRange(
174-
validator.getParameters().hasAttribute(EXTRACTOR_REALTIME_MODE_KEY)
175-
? EXTRACTOR_REALTIME_MODE_KEY
176-
: SOURCE_REALTIME_MODE_KEY,
177-
true,
178-
EXTRACTOR_REALTIME_MODE_FILE_VALUE,
179-
EXTRACTOR_REALTIME_MODE_HYBRID_VALUE,
180-
EXTRACTOR_REALTIME_MODE_LOG_VALUE,
181-
EXTRACTOR_REALTIME_MODE_FORCED_LOG_VALUE,
182-
EXTRACTOR_REALTIME_MODE_STREAM_MODE_VALUE,
183-
EXTRACTOR_REALTIME_MODE_BATCH_MODE_VALUE);
184-
}
185-
186184
// Validate source.start-time and source.end-time
187185
if (validator
188186
.getParameters()
@@ -260,6 +258,13 @@ private void constructRealtimeExtractor(final PipeParameters parameters) {
260258
return;
261259
}
262260

261+
if (!(pipeName != null
262+
&& (pipeName.startsWith(PipeStaticMeta.SUBSCRIPTION_PIPE_PREFIX)
263+
|| pipeName.startsWith(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)))) {
264+
realtimeExtractor = new PipeRealtimeDataRegionTsFileSource();
265+
return;
266+
}
267+
263268
// Use hybrid mode by default
264269
if (!parameters.hasAnyAttributes(EXTRACTOR_REALTIME_MODE_KEY, SOURCE_REALTIME_MODE_KEY)) {
265270
realtimeExtractor = new PipeRealtimeDataRegionHybridSource();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@
2424
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeNonCriticalException;
2525
import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
2626
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
27-
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
2827
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
2928
import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern;
3029
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
@@ -107,7 +106,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
107106
private final AtomicReference<Pair<Long, Long>> dataRegionTimePartitionIdBound =
108107
new AtomicReference<>();
109108

110-
protected boolean isForwardingPipeRequests;
109+
protected boolean isForwardingPipeRequests = true;
111110

112111
private boolean shouldTransferModFile; // Whether to transfer mods
113112

@@ -234,12 +233,7 @@ public void customize(
234233
? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime)
235234
: TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1;
236235

237-
isForwardingPipeRequests =
238-
parameters.getBooleanOrDefault(
239-
Arrays.asList(
240-
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_KEY,
241-
PipeSourceConstant.SOURCE_FORWARDING_PIPE_REQUESTS_KEY),
242-
PipeSourceConstant.EXTRACTOR_FORWARDING_PIPE_REQUESTS_DEFAULT_VALUE);
236+
isForwardingPipeRequests = true;
243237

244238
shouldTransferModFile =
245239
parameters.getBooleanOrDefault(

0 commit comments

Comments
 (0)