Skip to content

Pipe: disable unused logic #18129

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

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Pipe: disable unused logic #18129

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
45da56a
Pipe: disable unused logic in open source
Caideyipi Jul 7, 2026
2836daa
IT Fix
Caideyipi Jul 7, 2026
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
15 changes: 8 additions & 7 deletions ...src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeAlterIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ public void testBasicAlterPipe() throws Exception {
// The database & table name will be converted to lower case
final String sql =
String.format(
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='Test', 'table-name'='Test1', 'mode.streaming'='true') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='Test', 'table-name'='Test1', 'source.realtime.mode'='batch') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s')",
receiverDataNode.getIpAndPortString());
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
final Statement statement = connection.createStatement()) {
Expand All @@ -88,7 +88,7 @@ public void testBasicAlterPipe() throws Exception {
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=batch"));
Assert.assertTrue(
showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
Assert.assertTrue(
Expand Down Expand Up @@ -146,7 +146,7 @@ public void testBasicAlterPipe() throws Exception {
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test1"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test1"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=batch"));
Assert.assertTrue(
showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
Assert.assertTrue(
Expand Down Expand Up @@ -185,7 +185,8 @@ public void testBasicAlterPipe() throws Exception {
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("source=iotdb-source"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("database-name=test"));
Assert.assertTrue(showPipeResult.get(0).pipeExtractor.contains("table-name=test"));
Assert.assertFalse(showPipeResult.get(0).pipeExtractor.contains("mode.streaming=true"));
Assert.assertFalse(
showPipeResult.get(0).pipeExtractor.contains("source.realtime.mode=batch"));
Assert.assertTrue(
showPipeResult.get(0).pipeProcessor.contains("processor=do-nothing-processor"));
Assert.assertTrue(
Expand Down Expand Up @@ -443,7 +444,7 @@ public void testAlterPipeFailure() {
// Create pipe
sql =
String.format(
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test1', 'mode.streaming'='true') with sink ('node-urls'='%s', 'batch.enable'='false')",
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test1', 'source.realtime.mode'='batch') with sink ('node-urls'='%s', 'batch.enable'='false')",
receiverDataNode.getIpAndPortString());
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
final Statement statement = connection.createStatement()) {
Expand All @@ -468,7 +469,7 @@ public void testAlterPipeSourceAndSink() {
// Create pipe
final String sql =
String.format(
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test', 'mode.streaming'='true') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s', 'batch.enable'='false')",
"create pipe a2b with source ('source'='iotdb-source', 'database-name'='test', 'table-name'='test', 'source.realtime.mode'='batch') with processor ('processor'='do-nothing-processor') with sink ('node-urls'='%s', 'batch.enable'='false')",
receiverDataNode.getIpAndPortString());
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
final Statement statement = connection.createStatement()) {
Expand All @@ -488,7 +489,7 @@ public void testAlterPipeSourceAndSink() {
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
final Statement statement = connection.createStatement()) {
statement.execute(
"alter pipe a2b modify source('source' = 'iotdb-source','database-name'='test1', 'table-name'='test1', 'mode.streaming'='true', 'source.inclusion'='data.insert') modify sink ('batch.enable'='true')");
"alter pipe a2b modify source('source' = 'iotdb-source','database-name'='test1', 'table-name'='test1', 'source.realtime.mode'='batch', 'source.inclusion'='data.insert') modify sink ('batch.enable'='true')");
} catch (final SQLException e) {
fail(e.getMessage());
}
Expand Down
17 changes: 9 additions & 8 deletions .../test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeDataSinkIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {

TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");

TestUtils.executeNonQueries(
senderEnv,
Expand All @@ -87,7 +88,7 @@ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {
final Map<String, String> processorAttributes = new HashMap<>();
final Map<String, String> sinkAttributes = new HashMap<>();

sourceAttributes.put("source.realtime.mode", "log");
sourceAttributes.put("source.realtime.mode", "batch");
sourceAttributes.put("capture.table", "true");
sourceAttributes.put("__system.sql-dialect", "table");
sourceAttributes.put("capture.tree", "true");
Expand All @@ -112,6 +113,7 @@ public void testThriftConnectorWithRealtimeFirstDisabled() throws Exception {
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());

TableModelUtils.insertData("test", "test", 50, 100, senderEnv, true);
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");

TestUtils.executeNonQueries(
senderEnv,
Expand Down Expand Up @@ -144,11 +146,6 @@ public void testTsFileFormatAndAsyncLoad() throws Exception {
testSinkFormat("tsfile", true);
}

@Test
public void testSinkHybridFormat() throws Exception {
testSinkFormat("hybrid", false);
}

private void testSinkFormat(final String format, final boolean isAsyncLoad) throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

Expand All @@ -165,6 +162,7 @@ private void testSinkFormat(final String format, final boolean isAsyncLoad) thro

TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
TableModelUtils.insertData("test", "test", 0, 50, senderEnv, true);
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");

TestUtils.executeNonQueries(
senderEnv,
Expand Down Expand Up @@ -202,6 +200,7 @@ private void testSinkFormat(final String format, final boolean isAsyncLoad) thro
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());

TableModelUtils.insertData("test", "test", 50, 150, senderEnv, true);
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");

TestUtils.executeNonQueries(
senderEnv,
Expand Down Expand Up @@ -244,6 +243,7 @@ private void testSinkFormat(final String format, final boolean isAsyncLoad) thro
null);

TableModelUtils.insertData("test", "test", 150, 200, senderEnv, true);
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");
TableModelUtils.insertTablet("test", "test", 200, 250, senderEnv, true);
TableModelUtils.insertTablet("test", "test", 250, 300, senderEnv, true);
TableModelUtils.insertTablet("test", "test", 300, 350, senderEnv, true);
Expand Down Expand Up @@ -272,8 +272,8 @@ public void testWriteBackSink() throws Exception {
sourceAttributes.put("__system.sql-dialect", "table");
sourceAttributes.put("capture.tree", "true");
sourceAttributes.put("mode.double-living", "true");
sourceAttributes.put("forwarding-pipe-requests", "false");
sourceAttributes.put("source.database-name", "test.*");
sourceAttributes.put("source.realtime.mode", "batch");
sourceAttributes.put("source.database-name", "test");
sourceAttributes.put("source.table-name", "test.*");
sourceAttributes.put("user", "root");

Expand All @@ -299,6 +299,7 @@ public void testWriteBackSink() throws Exception {

TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
TableModelUtils.insertDataNotThrowError("test", "test", 0, 20, senderEnv);
TestUtils.executeNonQueryWithRetry(senderEnv, "flush");

TableModelUtils.insertTablet("test", "test", 20, 200, senderEnv, true);

Expand Down
58 changes: 3 additions & 55 deletions ...test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeLifeCycleIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,7 +204,7 @@ public void testLifeCycleLogMode() throws Exception {

sourceAttributes.put("capture.table", "true");
sourceAttributes.put("__system.sql-dialect", "table");
sourceAttributes.put("source.mode", "forced-log");
sourceAttributes.put("source.realtime.mode", "batch");
sourceAttributes.put("user", "root");

sinkAttributes.put("sink", "iotdb-thrift-sink");
Expand Down Expand Up @@ -261,7 +261,7 @@ public void testLifeCycleFileMode() throws Exception {

sourceAttributes.put("capture.table", "true");
sourceAttributes.put("__system.sql-dialect", "table");
sourceAttributes.put("mode.streaming", "false");
sourceAttributes.put("source.realtime.mode", "batch");
sourceAttributes.put("user", "root");

sinkAttributes.put("sink", "iotdb-thrift-sink");
Expand Down Expand Up @@ -292,59 +292,6 @@ public void testLifeCycleFileMode() throws Exception {
}
}

@Test
public void testLifeCycleHybridMode() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);

final String receiverIp = receiverDataNode.getIp();
final int receiverPort = receiverDataNode.getPort();
final Consumer<String> handleFailure =
o -> {
executeNonQueryWithRetry(senderEnv, "flush");
executeNonQueryWithRetry(receiverEnv, "flush");
};

try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {

TableModelUtils.createDataBaseAndTable(senderEnv, "test", "test");
TableModelUtils.insertData("test", "test", 0, 100, senderEnv);

final Map<String, String> sourceAttributes = new HashMap<>();
final Map<String, String> processorAttributes = new HashMap<>();
final Map<String, String> sinkAttributes = new HashMap<>();

sourceAttributes.put("capture.table", "true");
sourceAttributes.put("__system.sql-dialect", "table");
sourceAttributes.put("source.mode", "hybrid");
sourceAttributes.put("user", "root");

sinkAttributes.put("sink", "iotdb-thrift-sink");
sinkAttributes.put("sink.batch.enable", "false");
sinkAttributes.put("sink.ip", receiverIp);
sinkAttributes.put("sink.port", Integer.toString(receiverPort));

final TSStatus status =
client.createPipe(
new TCreatePipeReq("p1", sinkAttributes)
.setExtractorAttributes(sourceAttributes)
.setProcessorAttributes(processorAttributes));

Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());

Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("p1").getCode());

TableModelUtils.insertData("test", "test", 100, 200, senderEnv);

TableModelUtils.assertCountData("test", "test", 200, receiverEnv, handleFailure);

TableModelUtils.insertData("test", "test", 200, 300, senderEnv);

TableModelUtils.assertCountData("test", "test", 300, receiverEnv, handleFailure);
}
}

@Test
public void testLifeCycleWithClusterRestart() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
Expand Down Expand Up @@ -544,6 +491,7 @@ public void testReceiverAlreadyHaveTimeSeries() throws Exception {
}

@Test
@Ignore("Requires forwarding pipe request filtering, which is disabled.")
public void testDoubleLiving() throws Exception {

// Double living is two clusters with pipes connecting each other.
Expand Down
32 changes: 7 additions & 25 deletions .../test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,8 +219,7 @@ public void testPipeOnBothSenderAndReceiver() throws Exception {
extractorAttributes.put("database-name", "test");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("user", "root");

Expand Down Expand Up @@ -263,8 +262,7 @@ public void testPipeOnBothSenderAndReceiver() throws Exception {
extractorAttributes.put("database-name", "test.*");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("user", "root");

Expand Down Expand Up @@ -313,8 +311,7 @@ private void doTest() throws Exception {
extractorAttributes.put("database-name", "test.*");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("user", "root");

Expand Down Expand Up @@ -358,11 +355,6 @@ public void testAsyncConnectorUseNodeUrls() throws Exception {
doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName());
}

@Test
public void testAirGapConnectorUseNodeUrls() throws Exception {
doTestUseNodeUrls(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName());
}

private void doTestUseNodeUrls(String connectorName) throws Exception {
senderEnv
.getConfig()
Expand Down Expand Up @@ -407,16 +399,7 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {
};

for (final DataNodeWrapper wrapper : receiverEnv.getDataNodeWrapperList()) {
if (connectorName.equals(BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR.getPipePluginName())) {
// Use default port for convenience
nodeUrlsBuilder
.append(wrapper.getIp())
.append(":")
.append(wrapper.getPipeAirGapReceiverPort())
.append(",");
} else {
nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
}
nodeUrlsBuilder.append(wrapper.getIpAndPortString()).append(",");
}

try (final SyncConfigNodeIServiceClient client =
Expand All @@ -440,12 +423,11 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {
extractorAttributes.put("database-name", "test.*");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("user", "root");

// Test forced-log mode, in open releases this might be "file"
extractorAttributes.put("realtime.mode", "forced-log");
// Test batch mode
extractorAttributes.put("realtime.mode", "batch");

TSStatus status =
client.createPipe(
Expand All @@ -465,7 +447,7 @@ private void doTestUseNodeUrls(String connectorName) throws Exception {

TableModelUtils.assertData("test", "test", 0, 300, receiverEnv, handleFailure);

extractorAttributes.replace("realtime.mode", "file");
extractorAttributes.replace("realtime.mode", "batch");

status =
client.createPipe(
Expand Down
12 changes: 4 additions & 8 deletions ...t/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeSwitchStatusIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,8 +71,7 @@ public void testPipeSwitchStatus() throws Exception {
extractorAttributes.put("database-name", "test");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("start-time", "1");
extractorAttributes.put("end-time", "2");
Expand Down Expand Up @@ -170,8 +169,7 @@ public void testPipeIllegallySwitchStatus() throws Exception {
extractorAttributes.put("database-name", "test");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("start-time", "1");
extractorAttributes.put("end-time", "2");
Expand Down Expand Up @@ -257,8 +255,7 @@ public void testDropPipeAndCreateAgain() throws Exception {
extractorAttributes.put("database-name", "test");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("start-time", "0");
extractorAttributes.put("end-time", "200");
Expand Down Expand Up @@ -325,8 +322,7 @@ public void testWrongPipeName() throws Exception {
extractorAttributes.put("database-name", "test");
extractorAttributes.put("table-name", "test.*");
extractorAttributes.put("inclusion", "data.insert");
extractorAttributes.put("mode.streaming", "true");
extractorAttributes.put("mode.snapshot", "false");
extractorAttributes.put("source.realtime.mode", "batch");
extractorAttributes.put("mode.strict", "true");
extractorAttributes.put("start-time", "0");
extractorAttributes.put("end-time", "200");
Expand Down
Loading
Loading