Skip to content
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

Pipe: tree/table model isolation for alter/start/stop/drop/show pipe operations & support 'mode.double-living' #14386

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

Pipe: tree/table model isolation for alter/start/stop/drop/show pipe operations & support 'mode.double-living' #14386

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
2a1eacb
backup
VGalaxies Dec 11, 2024
abb453f
setup
VGalaxies Dec 11, 2024
53b249b
adapt DN
VGalaxies Dec 11, 2024
30b2faf
add IT
VGalaxies Dec 12, 2024
a023046
apply code review
VGalaxies Dec 17, 2024
fdb2d4a
minor improve
VGalaxies Dec 17, 2024
83fcc9d
basic support double living
VGalaxies Dec 17, 2024
e342640
support 'mode.double-living'
VGalaxies Dec 17, 2024
14cadf4
fix IT
VGalaxies Dec 17, 2024
ccf64de
improve double living & add IT
VGalaxies Dec 17, 2024
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
3 changes: 1 addition & 2 deletions integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT2AutoCreateSchema;
import org.apache.iotdb.itbase.env.BaseEnv;

import org.junit.Assert;
import org.junit.Test;
Expand Down Expand Up @@ -106,7 +105,7 @@ public void testBasicAlterPipe() throws Exception {
}

// Alter pipe (modify)
try (final Connection connection = senderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT);
try (final Connection connection = senderEnv.getConnection();
final Statement statement = connection.createStatement()) {
statement.execute("alter pipe a2b modify source ('source.pattern'='root.test2')");
} catch (SQLException e) {
Expand Down
23 changes: 15 additions & 8 deletions ...ation-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeSwitchStatusIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,11 +266,14 @@ public void testWrongPipeName() throws Exception {
.setExtractorAttributes(extractorAttributes)
.setProcessorAttributes(processorAttributes));
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("p0").getCode());
Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("p").getCode());
Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.startPipe("*").getCode());
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("p0").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("p").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.startPipe("*").getCode());
List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
Assert.assertTrue(
showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
Expand All @@ -281,10 +284,14 @@ public void testWrongPipeName() throws Exception {
Assert.assertTrue(
showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));

Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("").getCode());
Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("p0").getCode());
Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("p").getCode());
Assert.assertEquals(TSStatusCode.PIPE_ERROR.getStatusCode(), client.stopPipe("*").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("p0").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("p").getCode());
Assert.assertEquals(
TSStatusCode.PIPE_NOT_EXIST_ERROR.getStatusCode(), client.stopPipe("*").getCode());
showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
Assert.assertTrue(
showPipeResult.stream().anyMatch((o) -> o.id.equals("p1") && o.state.equals("RUNNING")));
Expand Down
30 changes: 20 additions & 10 deletions integration-test/src/test/java/org/apache/iotdb/pipe/it/tablemodel/IoTDBPipeAlterIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,8 @@ public void testBasicAlterPipe() throws Exception {
long lastCreationTime;
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// Check status
Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
Expand Down Expand Up @@ -96,7 +97,8 @@ public void testBasicAlterPipe() throws Exception {
// Show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// Check status
Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
Expand All @@ -114,7 +116,8 @@ public void testBasicAlterPipe() throws Exception {
// Show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// Check status
Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
Expand Down Expand Up @@ -149,7 +152,8 @@ public void testBasicAlterPipe() throws Exception {
// Show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// check status
Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
Expand Down Expand Up @@ -183,7 +187,8 @@ public void testBasicAlterPipe() throws Exception {
// Show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// Check status
Assert.assertEquals("STOPPED", showPipeResult.get(0).state);
Expand Down Expand Up @@ -225,7 +230,8 @@ public void testBasicAlterPipe() throws Exception {
// Show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// Check status
Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
Expand Down Expand Up @@ -257,7 +263,8 @@ public void testBasicAlterPipe() throws Exception {
// show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// check status
Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
Expand Down Expand Up @@ -289,7 +296,8 @@ public void testBasicAlterPipe() throws Exception {
// show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// check status
Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
Expand Down Expand Up @@ -323,7 +331,8 @@ public void testBasicAlterPipe() throws Exception {
// show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// check status
Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
Expand Down Expand Up @@ -354,7 +363,8 @@ public void testBasicAlterPipe() throws Exception {
// Show pipe
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
final List<TShowPipeInfo> showPipeResult = client.showPipe(new TShowPipeReq()).pipeInfoList;
final List<TShowPipeInfo> showPipeResult =
client.showPipe(new TShowPipeReq().setIsTableModel(true)).pipeInfoList;
Assert.assertEquals(1, showPipeResult.size());
// Check status
Assert.assertEquals("RUNNING", showPipeResult.get(0).state);
Expand Down
Loading
Loading