Skip to content

Commit 768e102

Browse files
committed
[Pipe] Preserve injected attributes in SHOW CREATE PIPE
1 parent 4c9ff8b commit 768e102

7 files changed

Lines changed: 1 addition & 50 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -365,7 +365,7 @@ public void testShowCreatePipe() throws SQLException {
365365
statement.executeQuery("show create pipe test_show_create_pipe"),
366366
"Pipe,Create Pipe,",
367367
Collections.singleton(
368-
"test_show_create_pipe,CREATE PIPE \"test_show_create_pipe\" WITH SINK ('sink'='do-nothing-sink'),"));
368+
"test_show_create_pipe,CREATE PIPE \"test_show_create_pipe\" WITH SINK ('sink'='do-nothing-sink','sink.enable-compaction'='true'),"));
369369
}
370370
}
371371

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/AlterPipeProcedureV2.java‎

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -134,16 +134,6 @@ public boolean executeFromValidateTask(final ConfigNodeProcedureEnv env) throws
134134
PipeSinkConstant.SINK_IOTDB_USERNAME_KEY,
135135
PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY,
136136
PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY);
137-
final boolean hasExplicitSinkCompaction =
138-
new PipeParameters(alterPipeRequest.getConnectorAttributes())
139-
.hasAnyAttributes(
140-
PipeSinkConstant.CONNECTOR_ENABLE_COMPACTION_KEY,
141-
PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY);
142-
if (hasExplicitSinkCompaction) {
143-
alterPipeRequest
144-
.getConnectorAttributes()
145-
.put(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY, Boolean.FALSE.toString());
146-
}
147137

148138
pipeTaskInfo.get().checkAndUpdateRequestBeforeAlterPipe(alterPipeRequest);
149139

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,6 @@ static TCreatePipeReq normalizeCreatePipeRequest(final TCreatePipeReq createPipe
128128
createPipeRequest
129129
.getConnectorAttributes()
130130
.put(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY, Boolean.TRUE.toString());
131-
createPipeRequest
132-
.getConnectorAttributes()
133-
.put(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY, Boolean.TRUE.toString());
134131
}
135132
return createPipeRequest;
136133
}

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2Test.java‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2323
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
2424
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
25-
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
2625
import org.apache.iotdb.confignode.manager.ConfigManager;
2726
import org.apache.iotdb.confignode.manager.PermissionManager;
2827
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
@@ -55,9 +54,6 @@ public void testNormalizeCreatePipeRequestEnablesCompactionForNewUserPipe() {
5554
assertEquals(
5655
Boolean.TRUE.toString(),
5756
request.getConnectorAttributes().get(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY));
58-
assertEquals(
59-
Boolean.TRUE.toString(),
60-
request.getConnectorAttributes().get(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY));
6157
}
6258

6359
@Test
@@ -75,10 +71,6 @@ public void testNormalizeCreatePipeRequestKeepsExplicitCompactionValue() {
7571
request.getConnectorAttributes().get(PipeSinkConstant.CONNECTOR_ENABLE_COMPACTION_KEY));
7672
assertFalse(
7773
request.getConnectorAttributes().containsKey(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY));
78-
assertFalse(
79-
request
80-
.getConnectorAttributes()
81-
.containsKey(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY));
8274
}
8375

8476
@Test

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowCreatePipeTask.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -143,12 +143,7 @@ private static Map<String, String> sanitizeSourceAttributes(final Map<String, St
143143
private static Map<String, String> sanitizeSinkAttributes(final Map<String, String> sink) {
144144
final boolean hasInjectedSinkAuthentication =
145145
Boolean.parseBoolean(sink.get(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY));
146-
final boolean hasInjectedSinkCompaction =
147-
Boolean.parseBoolean(sink.get(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY));
148146
final Map<String, String> result = sanitizeCommonAttributes(sink);
149-
if (hasInjectedSinkCompaction) {
150-
result.remove(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY);
151-
}
152147
result.remove(PipeSinkConstant.CONNECTOR_IOTDB_USER_ID);
153148
result.remove(PipeSinkConstant.SINK_IOTDB_USER_ID);
154149
result.remove(PipeSinkConstant.CONNECTOR_IOTDB_CLI_HOSTNAME);

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowCreateTaskTest.java‎

Lines changed: 0 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -93,15 +93,13 @@ public void testShowCreatePipeSQLShouldSanitizeInternalAndInjectedAttributes() {
9393

9494
final Map<String, String> sinkAttributes = new HashMap<>();
9595
sinkAttributes.put(PipeSinkConstant.SINK_KEY, "write-back-sink");
96-
sinkAttributes.put(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY, Boolean.TRUE.toString());
9796
sinkAttributes.put(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY, "true");
9897
sinkAttributes.put("__audit.sink", "audit");
9998
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USER_ID, "1");
10099
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USERNAME_KEY, "alice");
101100
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY, "hashed-password");
102101
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_CLI_HOSTNAME, "host");
103102
sinkAttributes.put(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY, Boolean.TRUE.toString());
104-
sinkAttributes.put(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY, Boolean.TRUE.toString());
105103

106104
final PipeMeta pipeMeta =
107105
new PipeMeta(
@@ -116,24 +114,6 @@ public void testShowCreatePipeSQLShouldSanitizeInternalAndInjectedAttributes() {
116114
ShowCreatePipeTask.getShowCreatePipeSQL(pipeMeta));
117115
}
118116

119-
@Test
120-
public void testShowCreatePipeSQLShouldKeepExplicitCompactionAttribute() {
121-
final Map<String, String> sinkAttributes = new HashMap<>();
122-
sinkAttributes.put(PipeSinkConstant.SINK_KEY, "do-nothing-sink");
123-
sinkAttributes.put(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY, Boolean.TRUE.toString());
124-
sinkAttributes.put(
125-
SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY, Boolean.FALSE.toString());
126-
127-
final PipeMeta pipeMeta =
128-
new PipeMeta(
129-
new PipeStaticMeta("test_pipe", 1L, new HashMap<>(), new HashMap<>(), sinkAttributes),
130-
new PipeRuntimeMeta());
131-
132-
assertEquals(
133-
"CREATE PIPE \"test_pipe\" WITH SINK ('sink'='do-nothing-sink','sink.enable-compaction'='true')",
134-
ShowCreatePipeTask.getShowCreatePipeSQL(pipeMeta));
135-
}
136-
137117
@Test
138118
public void testShowCreatePipeSQLShouldMaskExplicitCredentials() {
139119
final Map<String, String> sourceAttributes = new HashMap<>();

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/SystemConstant.java‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -48,8 +48,6 @@ public class SystemConstant {
4848
"__system.source-authentication-injected";
4949
public static final String SINK_AUTHENTICATION_INJECTED_KEY =
5050
"__system.sink-authentication-injected";
51-
public static final String SINK_ENABLE_COMPACTION_INJECTED_KEY =
52-
"__system.sink-enable-compaction-injected";
5351
public static final String SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY =
5452
"__system.sink-wait-for-schema-before-load";
5553

@@ -63,7 +61,6 @@ public class SystemConstant {
6361
SYSTEM_KEYS.add(PIPE_VISIBILITY_KEY);
6462
SYSTEM_KEYS.add(SOURCE_AUTHENTICATION_INJECTED_KEY);
6563
SYSTEM_KEYS.add(SINK_AUTHENTICATION_INJECTED_KEY);
66-
SYSTEM_KEYS.add(SINK_ENABLE_COMPACTION_INJECTED_KEY);
6764
SYSTEM_KEYS.add(SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY);
6865
}
6966

0 commit comments

Comments
 (0)