Skip to content

Pipe: serialize sink transfers by region #17946

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Jun 16, 2026
Merged

Pipe: serialize sink transfers by region #17946

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ...ava/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeSinkParallelIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ public void testIoTConnectorParallel() throws Exception {
connectorAttributes.put("connector.batch.enable", "true");
connectorAttributes.put("connector.ip", receiverIp);
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
connectorAttributes.put("connector.serialize-by-region", "false");
connectorAttributes.put("connector.parallel.tasks", "3");

final TSStatus status =
Expand Down
1 change: 1 addition & 0 deletions ...test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSinkParallelIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ public void testIoTConnectorParallel() throws Exception {
sinkAttributes.put("sink.batch.enable", "false");
sinkAttributes.put("sink.ip", receiverIp);
sinkAttributes.put("sink.port", Integer.toString(receiverPort));
sinkAttributes.put("sink.serialize-by-region", "false");
sinkAttributes.put("sink.parallel.tasks", "3");

final TSStatus status =
Expand Down
43 changes: 23 additions & 20 deletions ...rc/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
package org.apache.iotdb.db.pipe.agent.task.subtask.sink;

import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.task.connection.UnboundedBlockingPendingQueue;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
Expand Down Expand Up @@ -66,15 +65,14 @@

private final Map<String, String> attributeSortedString2DisplayString = new HashMap<>();

public synchronized String register(

Check warning on line 68 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 124 to 64, Complexity from 15 to 14, Nesting Level from 4 to 2, Number of Variables from 24 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ7OL--iK3HGsrH-eT8q&open=AZ7OL--iK3HGsrH-eT8q&pullRequest=17946
final Supplier<? extends PipeSinkSubtaskExecutor> executorSupplier,
final PipeParameters pipeSinkParameters,
final PipeTaskSinkRuntimeEnvironment environment) {
final String connectorName =
PipeSinkConstant.getConnectorOrSinkNameWithDefault(pipeSinkParameters);
final String connectorKey =
pipeSinkParameters
.getStringOrDefault(
Arrays.asList(PipeSinkConstant.CONNECTOR_KEY, PipeSinkConstant.SINK_KEY),
BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
connectorName
// Convert the value of `CONNECTOR_KEY` or `SINK_KEY` to lowercase
// for matching in `CONNECTOR_CONSTRUCTORS`
.toLowerCase();
Expand All @@ -93,38 +91,43 @@

final int sinkNum;
boolean realTimeFirst = false;
boolean serializeByRegion = false;
String attributeSortedString = generateAttributeSortedString(pipeSinkParameters);
final String attributeDisplayString = generateAttributeDisplayString(pipeSinkParameters);
if (isDataRegionSink) {
serializeByRegion = PipeSinkConstant.isSerializeByRegionEnabled(pipeSinkParameters);
sinkNum =
pipeSinkParameters.getIntOrDefault(
Arrays.asList(
PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
PipeSinkConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
PipeSinkConstant.SINGLE_THREAD_DEFAULT_SINK.contains(
pipeSinkParameters
.getStringOrDefault(
Arrays.asList(
PipeSinkConstant.CONNECTOR_KEY, PipeSinkConstant.SINK_KEY),
BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName())
.toLowerCase())
? 1
: PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);
serializeByRegion
? 1
: pipeSinkParameters.getIntOrDefault(
Arrays.asList(
PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_KEY,
PipeSinkConstant.SINK_IOTDB_PARALLEL_TASKS_KEY),
PipeSinkConstant.SINGLE_THREAD_DEFAULT_SINK.contains(connectorKey)
? 1
: PipeSinkConstant.CONNECTOR_IOTDB_PARALLEL_TASKS_DEFAULT_VALUE);

Check warning on line 108 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Extract this nested ternary operation into an independent statement. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ7OZO8QbH8oLnz3hdMf&open=AZ7OZO8QbH8oLnz3hdMf&pullRequest=17946
realTimeFirst =
pipeSinkParameters.getBooleanOrDefault(
Arrays.asList(
PipeSinkConstant.CONNECTOR_REALTIME_FIRST_KEY,
PipeSinkConstant.SINK_REALTIME_FIRST_KEY),
PipeSinkConstant.CONNECTOR_REALTIME_FIRST_DEFAULT_VALUE);
attributeSortedString = "data_" + attributeSortedString;
attributeSortedString =
serializeByRegion
? "data_region_" + environment.getRegionId() + "_" + attributeSortedString
: "data_" + attributeSortedString;
} else {
// Do not allow parallel tasks for schema region connectors
// to avoid the potential disorder of the schema region data transfer
sinkNum = 1;
attributeSortedString = "schema_" + attributeSortedString;
}
final String attributeDisplayStringWithPrefix =
isDataRegionSink ? "data_" + attributeDisplayString : "schema_" + attributeDisplayString;
isDataRegionSink
? serializeByRegion
? "data_region_" + environment.getRegionId() + "_" + attributeDisplayString
: "data_" + attributeDisplayString

Check warning on line 129 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskManager.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Extract this nested ternary operation into an independent statement. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ7OL--iK3HGsrH-eT8p&open=AZ7OL--iK3HGsrH-eT8p&pullRequest=17946
: "schema_" + attributeDisplayString;
environment.setAttributeSortedString(attributeDisplayStringWithPrefix);

if (!attributeSortedString2SubtaskLifeCycleMap.containsKey(attributeSortedString)) {
Expand Down
16 changes: 16 additions & 0 deletions ...node/src/test/java/org/apache/iotdb/db/pipe/agent/plugin/PipeDataNodePluginAgentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor;
import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink;
import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
import org.apache.iotdb.db.pipe.sink.protocol.thrift.sync.IoTDBDataRegionSyncSink;
import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;

Expand Down Expand Up @@ -148,6 +149,21 @@ public void testPipePluginAgent() {
}
}))
.getClass());
Assert.assertEquals(
IoTDBDataRegionSyncSink.class,
agent.dataRegion().reflectSink(new PipeParameters(new HashMap<>())).getClass());
Assert.assertEquals(
IoTDBDataRegionAsyncSink.class,
agent
.dataRegion()
.reflectSink(
new PipeParameters(
new HashMap<String, String>() {
{
put(PipeSinkConstant.CONNECTOR_SERIALIZE_BY_REGION_KEY, "false");
}
}))
.getClass());
Assert.assertEquals(
IoTConsensusV2AsyncSink.class,
agent
Expand Down
8 changes: 1 addition & 7 deletions ...main/java/org/apache/iotdb/commons/pipe/agent/plugin/constructor/PipeSinkConstructor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,11 @@

package org.apache.iotdb.commons.pipe.agent.plugin.constructor;

import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMetaKeeper;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;

import java.util.Arrays;

public abstract class PipeSinkConstructor extends PipePluginConstructor {

protected PipeSinkConstructor(PipePluginMetaKeeper pipePluginMetaKeeper) {
Expand All @@ -41,10 +38,7 @@ protected PipeSinkConstructor() {
public final PipeConnector reflectPlugin(PipeParameters connectorParameters) {
return (PipeConnector)
reflectPluginByKey(
connectorParameters
.getStringOrDefault(
Arrays.asList(PipeSinkConstant.CONNECTOR_KEY, PipeSinkConstant.SINK_KEY),
BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName())
PipeSinkConstant.getConnectorOrSinkNameWithDefault(connectorParameters)
// Convert the value of `CONNECTOR_KEY` or `SINK_KEY` to lowercase for matching in
// `PLUGIN_CONSTRUCTORS`
.toLowerCase());
Expand Down
22 changes: 22 additions & 0 deletions ...commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.iotdb.commons.i18n.PipeMessages;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;

import com.github.luben.zstd.Zstd;

Expand Down Expand Up @@ -69,6 +70,27 @@ public class PipeSinkConstant {
public static final String SINK_REALTIME_FIRST_KEY = "sink.realtime-first";
public static final boolean CONNECTOR_REALTIME_FIRST_DEFAULT_VALUE = true;

public static final String CONNECTOR_SERIALIZE_BY_REGION_KEY = "connector.serialize-by-region";
public static final String SINK_SERIALIZE_BY_REGION_KEY = "sink.serialize-by-region";
public static final boolean CONNECTOR_SERIALIZE_BY_REGION_DEFAULT_VALUE = true;

public static boolean isSerializeByRegionEnabled(final PipeParameters parameters) {
return parameters.getBooleanOrDefault(
Arrays.asList(CONNECTOR_SERIALIZE_BY_REGION_KEY, SINK_SERIALIZE_BY_REGION_KEY),
CONNECTOR_SERIALIZE_BY_REGION_DEFAULT_VALUE);
}

public static String getConnectorOrSinkNameWithDefault(final PipeParameters parameters) {
return parameters.getStringOrDefault(
Arrays.asList(CONNECTOR_KEY, SINK_KEY), getDefaultConnectorOrSinkName(parameters));
}

private static String getDefaultConnectorOrSinkName(final PipeParameters parameters) {
return isSerializeByRegionEnabled(parameters)
? BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName()
: BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName();
}

public static final String CONNECTOR_IOTDB_BATCH_MODE_ENABLE_KEY = "connector.batch.enable";
public static final String SINK_IOTDB_BATCH_MODE_ENABLE_KEY = "sink.batch.enable";
public static final boolean CONNECTOR_IOTDB_BATCH_MODE_ENABLE_DEFAULT_VALUE = true;
Expand Down
Loading