Skip to content

Commit 6348c2d

Browse files
CaideyipiCaideyipi
authored
[to dev/1.3] Pipe: backport auto split fixes (#17866)
* Pipe: Fixed the bug that separated historical pipe may not include mod on deletion & The pipe without data.insert may be wrongly separated by pipe and transfer data (#17346) * fix * fix * f (cherry picked from commit 355a872) * Pipe: Made the historical pipe split auto dropped after completion (#17295) * snapshot * may-comp * auto (cherry picked from commit 9bfe0b0)
1 parent 78a4790 commit 6348c2d

4 files changed

Lines changed: 138 additions & 78 deletions

File tree

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

Lines changed: 51 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,10 @@
2121

2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2323
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
24+
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
2425
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
2526
import org.apache.iotdb.db.it.utils.TestUtils;
27+
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
2628
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
2729
import org.apache.iotdb.it.framework.IoTDBTestRunner;
2830
import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema;
@@ -33,59 +35,67 @@
3335
import org.junit.experimental.categories.Category;
3436
import org.junit.runner.RunWith;
3537

38+
import java.sql.Connection;
39+
import java.sql.ResultSet;
40+
import java.sql.Statement;
3641
import java.util.Collections;
3742
import java.util.HashMap;
3843
import java.util.Map;
44+
import java.util.concurrent.TimeUnit;
45+
46+
import static org.awaitility.Awaitility.await;
3947

4048
@RunWith(IoTDBTestRunner.class)
4149
@Category({MultiClusterIT2AutoCreateSchema.class})
4250
public class IoTDBPipeAutoDropIT extends AbstractPipeDualAutoIT {
4351

52+
@Override
53+
protected void setupConfig() {
54+
super.setupConfig();
55+
senderEnv.getConfig().getCommonConfig().setPipeAutoSplitFullEnabled(true);
56+
receiverEnv.getConfig().getCommonConfig().setPipeAutoSplitFullEnabled(true);
57+
senderEnv.getConfig().getConfigNodeConfig().setLeaderDistributionPolicy("HASH");
58+
}
59+
4460
@Test
4561
public void testAutoDropInHistoricalTransfer() throws Exception {
4662
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
4763

48-
final String receiverIp = receiverDataNode.getIp();
49-
final int receiverPort = receiverDataNode.getPort();
50-
51-
try (final SyncConfigNodeIServiceClient client =
52-
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
53-
54-
TestUtils.executeNonQuery(senderEnv, "insert into root.db.d1(time,s1) values (1,1)", null);
55-
56-
final Map<String, String> extractorAttributes = new HashMap<>();
57-
final Map<String, String> processorAttributes = new HashMap<>();
58-
final Map<String, String> connectorAttributes = new HashMap<>();
59-
60-
extractorAttributes.put("extractor.mode", "query");
61-
62-
connectorAttributes.put("connector", "iotdb-thrift-connector");
63-
connectorAttributes.put("connector.batch.enable", "false");
64-
connectorAttributes.put("connector.ip", receiverIp);
65-
connectorAttributes.put("connector.port", Integer.toString(receiverPort));
66-
67-
final TSStatus status =
68-
client.createPipe(
69-
new TCreatePipeReq("p1", connectorAttributes)
70-
.setExtractorAttributes(extractorAttributes)
71-
.setProcessorAttributes(processorAttributes));
72-
73-
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
74-
Assert.assertEquals(
75-
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());
76-
77-
TestUtils.assertDataEventuallyOnEnv(
78-
receiverEnv,
79-
"select count(*) from root.**",
80-
"count(root.db.d1.s1),",
81-
Collections.singleton("1,"));
82-
83-
TestUtils.assertDataEventuallyOnEnv(
84-
senderEnv,
85-
"show pipes",
86-
"ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,",
87-
Collections.emptySet());
88-
}
64+
TestUtils.executeNonQuery(
65+
senderEnv,
66+
String.format(
67+
"create pipe a2b with sink ('node-urls'='%s')", receiverDataNode.getIpAndPortString()),
68+
null);
69+
70+
TestUtils.executeNonQuery(senderEnv, "insert into root.db.d1(time,s1) values (1,1)", null);
71+
72+
TestUtils.assertDataEventuallyOnEnv(
73+
receiverEnv,
74+
"select count(*) from root.**",
75+
"count(root.db.d1.s1),",
76+
Collections.singleton("1,"));
77+
78+
await()
79+
.pollInSameThread()
80+
.pollDelay(1L, TimeUnit.SECONDS)
81+
.pollInterval(1L, TimeUnit.SECONDS)
82+
.atMost(600, TimeUnit.SECONDS)
83+
.untilAsserted(
84+
() -> {
85+
try (final Connection connection = senderEnv.getConnection();
86+
final Statement statement = connection.createStatement();
87+
final ResultSet result = statement.executeQuery("show pipes")) {
88+
int pipeNum = 0;
89+
while (result.next()) {
90+
final String pipeName = result.getString(ColumnHeaderConstant.ID);
91+
if (!pipeName.contains(PipeStaticMeta.CONSENSUS_PIPE_PREFIX)
92+
&& pipeName.endsWith("_history")) {
93+
pipeNum++;
94+
}
95+
}
96+
Assert.assertEquals(0, pipeNum);
97+
}
98+
});
8999
}
90100

91101
@Test

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

Lines changed: 50 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeInfo;
2424
import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq;
2525
import org.apache.iotdb.consensus.ConsensusFactory;
26+
import org.apache.iotdb.db.it.utils.TestUtils;
2627
import org.apache.iotdb.it.env.MultiEnvFactory;
2728
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
2829
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -37,6 +38,8 @@
3738
import java.sql.Connection;
3839
import java.sql.SQLException;
3940
import java.sql.Statement;
41+
import java.util.Arrays;
42+
import java.util.Collections;
4043
import java.util.List;
4144
import java.util.Objects;
4245

@@ -78,13 +81,12 @@ public void setUp() {
7881
public void testSingleEnv() throws Exception {
7982
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
8083

81-
final String sql =
82-
String.format(
83-
"create pipe a2b with source ('source'='iotdb-source') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
84-
receiverDataNode.getIpAndPortString());
8584
try (final Connection connection = senderEnv.getConnection();
8685
final Statement statement = connection.createStatement()) {
87-
statement.execute(sql);
86+
statement.execute(
87+
String.format(
88+
"create pipe a2b with sink ('node-urls'='%s')",
89+
receiverDataNode.getIpAndPortString()));
8890
} catch (final SQLException e) {
8991
fail(e.getMessage());
9092
}
@@ -100,5 +102,48 @@ public void testSingleEnv() throws Exception {
100102
|| (Objects.equals(showPipeResult.get(1).id, "a2b_history")
101103
&& Objects.equals(showPipeResult.get(0).id, "a2b_realtime")));
102104
}
105+
106+
// Do not split for pipes without insertion or non-full
107+
TestUtils.executeNonQueries(
108+
senderEnv,
109+
Arrays.asList(
110+
"drop pipe a2b_history",
111+
"drop pipe a2b_realtime",
112+
String.format(
113+
"create pipe a2b1 with source ('inclusion'='schema') with sink ('node-urls'='%s')",
114+
receiverDataNode.getIpAndPortString()),
115+
String.format(
116+
"create pipe a2b2 with source ('realtime.enable'='false') with sink ('node-urls'='%s')",
117+
receiverDataNode.getIpAndPortString()),
118+
String.format(
119+
"create pipe a2b3 with source ('history.enable'='false') with sink ('node-urls'='%s')",
120+
receiverDataNode.getIpAndPortString())),
121+
null);
122+
123+
try (final SyncConfigNodeIServiceClient client =
124+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
125+
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
126+
showPipeResult.removeIf(i -> i.getId().startsWith("__consensus"));
127+
Assert.assertEquals(3, showPipeResult.size());
128+
}
129+
130+
TestUtils.executeNonQueries(
131+
senderEnv,
132+
Arrays.asList(
133+
"drop pipe a2b1",
134+
"drop pipe a2b2",
135+
"drop pipe a2b3",
136+
"insert into root.test.device(time, field) values(0,1),(1,2)",
137+
"delete from root.test.device.* where time == 0",
138+
String.format(
139+
"create pipe a2b with source ('inclusion'='all') with sink ('node-urls'='%s')",
140+
receiverDataNode.getIpAndPortString())),
141+
null);
142+
143+
TestUtils.assertDataEventuallyOnEnv(
144+
receiverEnv,
145+
"select * from root.test.device",
146+
"Time,root.test.device.field,",
147+
Collections.singleton("1,2.0,"));
103148
}
104149
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -586,7 +586,7 @@ public boolean hasPipeReleaseRegionRelatedResource(final int consensusGroupId) {
586586
}
587587
}
588588

589-
public boolean isFullSync(final PipeParameters parameters) {
589+
public boolean isFullSync(final PipeParameters parameters) throws IllegalPathException {
590590
if (isSnapshotMode(parameters)) {
591591
return false;
592592
}
@@ -600,7 +600,10 @@ public boolean isFullSync(final PipeParameters parameters) {
600600
Arrays.asList(EXTRACTOR_REALTIME_ENABLE_KEY, SOURCE_REALTIME_ENABLE_KEY),
601601
EXTRACTOR_REALTIME_ENABLE_DEFAULT_VALUE);
602602

603-
return isHistoryEnable && isRealtimeEnable;
603+
return isHistoryEnable
604+
&& isRealtimeEnable
605+
&& DataRegionListeningFilter.parseInsertionDeletionListeningOptionPair(parameters)
606+
.getLeft();
604607
}
605608

606609
@Override

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java‎

Lines changed: 32 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@
140140
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
141141
import org.apache.iotdb.db.exception.sql.SemanticException;
142142
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
143+
import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
143144
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
144145
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
145146
import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
@@ -1826,10 +1827,10 @@ public SettableFuture<ConfigTaskResult> createPipe(
18261827
new PipeParameters(createPipeStatement.getSourceAttributes());
18271828
final PipeParameters sinkPipeParameters =
18281829
new PipeParameters(createPipeStatement.getSinkAttributes());
1829-
if (PipeConfig.getInstance().getPipeAutoSplitFullEnabled()
1830-
&& PipeDataNodeAgent.task().isFullSync(sourcePipeParameters)) {
1831-
try (final ConfigNodeClient configNodeClient =
1832-
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
1830+
try (final ConfigNodeClient configNodeClient =
1831+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
1832+
if (PipeConfig.getInstance().getPipeAutoSplitFullEnabled()
1833+
&& PipeDataNodeAgent.task().isFullSync(sourcePipeParameters)) {
18331834
// 1. Send request to create the real-time data synchronization pipeline
18341835
final TCreatePipeReq realtimeReq =
18351836
new TCreatePipeReq()
@@ -1874,11 +1875,19 @@ public SettableFuture<ConfigTaskResult> createPipe(
18741875
Boolean.toString(false),
18751876
PipeSourceConstant.EXTRACTOR_HISTORY_ENABLE_KEY,
18761877
Boolean.toString(true),
1877-
// We force the historical pipe to transfer data only
1878+
PipeSourceConstant.EXTRACTOR_MODE_KEY,
1879+
PipeSourceConstant.EXTRACTOR_MODE_SNAPSHOT_VALUE,
1880+
// We force the historical pipe to transfer data (and maybe
1881+
// deletion) only
18781882
// Thus we can transfer schema only once
18791883
// And may drop the historical pipe on successfully transferred
18801884
PipeSourceConstant.SOURCE_INCLUSION_KEY,
1881-
PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE,
1885+
DataRegionListeningFilter
1886+
.parseInsertionDeletionListeningOptionPair(
1887+
sourcePipeParameters)
1888+
.getRight()
1889+
? "data"
1890+
: PipeSourceConstant.EXTRACTOR_INCLUSION_DEFAULT_VALUE,
18821891
PipeSourceConstant.SOURCE_EXCLUSION_KEY,
18831892
PipeSourceConstant.EXTRACTOR_EXCLUSION_DEFAULT_VALUE)))
18841893
.getAttribute())
@@ -1901,31 +1910,24 @@ public SettableFuture<ConfigTaskResult> createPipe(
19011910

19021911
// 3. Set success status only if both pipelines are created successfully
19031912
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
1904-
} catch (final Exception e) {
1905-
// Catch any other exceptions (e.g., network issues)
1906-
future.setException(e);
1907-
}
1908-
return future;
1909-
}
1910-
1911-
try (final ConfigNodeClient configNodeClient =
1912-
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
1913-
TCreatePipeReq req =
1914-
new TCreatePipeReq()
1915-
.setPipeName(pipeName)
1916-
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
1917-
.setExtractorAttributes(createPipeStatement.getSourceAttributes())
1918-
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
1919-
.setConnectorAttributes(createPipeStatement.getSinkAttributes());
1920-
TSStatus tsStatus = configNodeClient.createPipe(req);
1921-
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
1922-
LOGGER.warn(
1923-
"Failed to create pipe {} in config node, status is {}.",
1924-
createPipeStatement.getPipeName(),
1925-
tsStatus);
1926-
future.setException(new IoTDBException(tsStatus));
19271913
} else {
1928-
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
1914+
final TCreatePipeReq req =
1915+
new TCreatePipeReq()
1916+
.setPipeName(pipeName)
1917+
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
1918+
.setExtractorAttributes(createPipeStatement.getSourceAttributes())
1919+
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
1920+
.setConnectorAttributes(createPipeStatement.getSinkAttributes());
1921+
final TSStatus tsStatus = configNodeClient.createPipe(req);
1922+
if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
1923+
LOGGER.warn(
1924+
"Failed to create pipe {} in config node, status is {}.",
1925+
createPipeStatement.getPipeName(),
1926+
tsStatus);
1927+
future.setException(new IoTDBException(tsStatus));
1928+
} else {
1929+
future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
1930+
}
19291931
}
19301932
} catch (final Exception e) {
19311933
future.setException(e);

0 commit comments

Comments
 (0)