Skip to content

Commit 4c9ff8b

Browse files
committed
Fix SHOW CREATE PIPE compaction defaults
1 parent d23e21a commit 4c9ff8b

6 files changed

Lines changed: 49 additions & 0 deletions

File tree

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,16 @@ 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+
}
137147

138148
pipeTaskInfo.get().checkAndUpdateRequestBeforeAlterPipe(alterPipeRequest);
139149

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,9 @@ 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());
131134
}
132135
return createPipeRequest;
133136
}

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
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;
2526
import org.apache.iotdb.confignode.manager.ConfigManager;
2627
import org.apache.iotdb.confignode.manager.PermissionManager;
2728
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
@@ -54,6 +55,9 @@ public void testNormalizeCreatePipeRequestEnablesCompactionForNewUserPipe() {
5455
assertEquals(
5556
Boolean.TRUE.toString(),
5657
request.getConnectorAttributes().get(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY));
58+
assertEquals(
59+
Boolean.TRUE.toString(),
60+
request.getConnectorAttributes().get(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY));
5761
}
5862

5963
@Test
@@ -71,6 +75,10 @@ public void testNormalizeCreatePipeRequestKeepsExplicitCompactionValue() {
7175
request.getConnectorAttributes().get(PipeSinkConstant.CONNECTOR_ENABLE_COMPACTION_KEY));
7276
assertFalse(
7377
request.getConnectorAttributes().containsKey(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY));
78+
assertFalse(
79+
request
80+
.getConnectorAttributes()
81+
.containsKey(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY));
7482
}
7583

7684
@Test

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,12 @@ 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));
146148
final Map<String, String> result = sanitizeCommonAttributes(sink);
149+
if (hasInjectedSinkCompaction) {
150+
result.remove(PipeSinkConstant.SINK_ENABLE_COMPACTION_KEY);
151+
}
147152
result.remove(PipeSinkConstant.CONNECTOR_IOTDB_USER_ID);
148153
result.remove(PipeSinkConstant.SINK_IOTDB_USER_ID);
149154
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: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,13 +93,15 @@ 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());
9697
sinkAttributes.put(SystemConstant.RESTART_OR_NEWLY_ADDED_KEY, "true");
9798
sinkAttributes.put("__audit.sink", "audit");
9899
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USER_ID, "1");
99100
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_USERNAME_KEY, "alice");
100101
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY, "hashed-password");
101102
sinkAttributes.put(PipeSinkConstant.SINK_IOTDB_CLI_HOSTNAME, "host");
102103
sinkAttributes.put(SystemConstant.SINK_AUTHENTICATION_INJECTED_KEY, Boolean.TRUE.toString());
104+
sinkAttributes.put(SystemConstant.SINK_ENABLE_COMPACTION_INJECTED_KEY, Boolean.TRUE.toString());
103105

104106
final PipeMeta pipeMeta =
105107
new PipeMeta(
@@ -114,6 +116,24 @@ public void testShowCreatePipeSQLShouldSanitizeInternalAndInjectedAttributes() {
114116
ShowCreatePipeTask.getShowCreatePipeSQL(pipeMeta));
115117
}
116118

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+
117137
@Test
118138
public void testShowCreatePipeSQLShouldMaskExplicitCredentials() {
119139
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: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,8 @@ 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";
5153
public static final String SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY =
5254
"__system.sink-wait-for-schema-before-load";
5355

@@ -61,6 +63,7 @@ public class SystemConstant {
6163
SYSTEM_KEYS.add(PIPE_VISIBILITY_KEY);
6264
SYSTEM_KEYS.add(SOURCE_AUTHENTICATION_INJECTED_KEY);
6365
SYSTEM_KEYS.add(SINK_AUTHENTICATION_INJECTED_KEY);
66+
SYSTEM_KEYS.add(SINK_ENABLE_COMPACTION_INJECTED_KEY);
6467
SYSTEM_KEYS.add(SINK_WAIT_FOR_SCHEMA_BEFORE_LOAD_KEY);
6568
}
6669

0 commit comments

Comments
 (0)