Use consensus read in show table
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
index 00e7345..49dcc2b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
@@ -41,6 +41,7 @@
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan;
import org.apache.iotdb.confignode.consensus.request.read.subscription.ShowSubscriptionPlan;
import org.apache.iotdb.confignode.consensus.request.read.subscription.ShowTopicPlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.CheckTemplateSettablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetAllSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetAllTemplateSetInfoPlan;
@@ -416,6 +417,9 @@
case SetTableProperties:
plan = new SetTablePropertiesPlan();
break;
+ case ShowTable:
+ plan = new ShowTablePlan();
+ break;
case GetNodePathsPartition:
plan = new GetNodePathsPartitionPlan();
break;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
index 8defa85..3b011e5 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
@@ -160,6 +160,7 @@
CommitCreateTable((short) 852),
AddTableColumn((short) 853),
SetTableProperties((short) 854),
+ ShowTable((short) 855),
/** Deprecated types for sync, restored them for upgrade. */
@Deprecated
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/plugin/GetPipePluginJarPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/plugin/GetPipePluginJarPlan.java
index 4b5b215..363b88a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/plugin/GetPipePluginJarPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/plugin/GetPipePluginJarPlan.java
@@ -37,7 +37,7 @@
super(ConfigPhysicalPlanType.GetPipePluginJar);
}
- public GetPipePluginJarPlan(List<String> jarNames) {
+ public GetPipePluginJarPlan(final List<String> jarNames) {
super(ConfigPhysicalPlanType.GetPipePluginJar);
this.jarNames = jarNames;
}
@@ -47,18 +47,18 @@
}
@Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
+ protected void serializeImpl(final DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
ReadWriteIOUtils.write(jarNames.size(), stream);
- for (String jarName : jarNames) {
+ for (final String jarName : jarNames) {
ReadWriteIOUtils.write(jarName, stream);
}
}
@Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
- int size = ReadWriteIOUtils.readInt(buffer);
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
+ final int size = ReadWriteIOUtils.readInt(buffer);
jarNames = new ArrayList<>();
for (int i = 0; i < size; i++) {
jarNames.add(ReadWriteIOUtils.readString(buffer));
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/task/ShowPipePlanV2.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/task/ShowPipePlanV2.java
index 1d1acf2..0cd6d9e 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/task/ShowPipePlanV2.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/pipe/task/ShowPipePlanV2.java
@@ -33,12 +33,12 @@
}
@Override
- protected void serializeImpl(DataOutputStream stream) throws IOException {
+ protected void serializeImpl(final DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
}
@Override
- protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
// Empty method, since it is not needed now
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/table/ShowTablePlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/table/ShowTablePlan.java
new file mode 100644
index 0000000..48d3b4a
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/table/ShowTablePlan.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.consensus.request.read.table;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class ShowTablePlan extends ConfigPhysicalPlan {
+
+ private String database;
+
+ public ShowTablePlan() {
+ super(ConfigPhysicalPlanType.ShowTable);
+ }
+
+ public ShowTablePlan(final String database) {
+ super(ConfigPhysicalPlanType.ShowTable);
+ this.database = database;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ @Override
+ protected void serializeImpl(final DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ReadWriteIOUtils.write(database, stream);
+ }
+
+ @Override
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
+ database = ReadWriteIOUtils.readString(buffer);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
index 4cbe230..4085f30 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java
@@ -42,6 +42,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;
@@ -61,39 +62,34 @@
}
public PipeTableResp filter(final Boolean whereClause, final String pipeName) {
+ if (Objects.isNull(pipeName)) {
+ return this;
+ }
if (whereClause == null || !whereClause) {
- if (pipeName == null) {
- return this;
- } else {
- return new PipeTableResp(
- status,
- allPipeMeta.stream()
- .filter(pipeMeta -> pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
- .collect(Collectors.toList()));
- }
+ return new PipeTableResp(
+ status,
+ allPipeMeta.stream()
+ .filter(pipeMeta -> pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
+ .collect(Collectors.toList()));
} else {
- if (pipeName == null) {
- return this;
- } else {
- final String sortedConnectorParametersString =
- allPipeMeta.stream()
- .filter(pipeMeta -> pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
- .findFirst()
- .map(pipeMeta -> pipeMeta.getStaticMeta().getConnectorParameters().toString())
- .orElse(null);
+ final String sortedConnectorParametersString =
+ allPipeMeta.stream()
+ .filter(pipeMeta -> pipeMeta.getStaticMeta().getPipeName().equals(pipeName))
+ .findFirst()
+ .map(pipeMeta -> pipeMeta.getStaticMeta().getConnectorParameters().toString())
+ .orElse(null);
- return new PipeTableResp(
- status,
- allPipeMeta.stream()
- .filter(
- pipeMeta ->
- pipeMeta
- .getStaticMeta()
- .getConnectorParameters()
- .toString()
- .equals(sortedConnectorParametersString))
- .collect(Collectors.toList()));
- }
+ return new PipeTableResp(
+ status,
+ allPipeMeta.stream()
+ .filter(
+ pipeMeta ->
+ pipeMeta
+ .getStaticMeta()
+ .getConnectorParameters()
+ .toString()
+ .equals(sortedConnectorParametersString))
+ .collect(Collectors.toList()));
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/table/ShowTableResp.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/table/ShowTableResp.java
new file mode 100644
index 0000000..96cef09
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/table/ShowTableResp.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.consensus.response.table;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp;
+import org.apache.iotdb.confignode.rpc.thrift.TTableInfo;
+import org.apache.iotdb.consensus.common.DataSet;
+
+import java.util.List;
+
+public class ShowTableResp implements DataSet {
+ private final TSStatus status;
+ private final List<TTableInfo> tableInfoList;
+
+ public ShowTableResp(final TSStatus status, final List<TTableInfo> tableInfoList) {
+ this.status = status;
+ this.tableInfoList = tableInfoList;
+ }
+
+ public TShowTableResp convertToTShowTableResp() {
+ return new TShowTableResp(status).setTableInfoList(tableInfoList);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 0b5616f..36c642f 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -2576,11 +2576,9 @@
@Override
public TShowTableResp showTables(final String database) {
final TSStatus status = confirmLeader();
- if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return clusterSchemaManager.showTables(database);
- } else {
- return new TShowTableResp(status);
- }
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ ? clusterSchemaManager.showTables(database)
+ : new TShowTableResp(status);
}
@Override
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
index 100da33..0d370bb 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
@@ -181,14 +181,14 @@
"Failed to drop pipe %s. Failures: %s does not exist.", pipeName, pipeName));
}
- public TShowPipeResp showPipes(TShowPipeReq req) {
+ public TShowPipeResp showPipes(final TShowPipeReq req) {
try {
return ((PipeTableResp) configManager.getConsensusManager().read(new ShowPipePlanV2()))
.filter(req.whereClause, req.pipeName)
.convertToTShowPipeResp();
- } catch (ConsensusException e) {
+ } catch (final ConsensusException e) {
LOGGER.warn("Failed in the read API executing the consensus layer due to: ", e);
- TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ final TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
res.setMessage(e.getMessage());
return new PipeTableResp(res, Collections.emptyList()).convertToTShowPipeResp();
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
index b6419ff..15422f5 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -39,6 +39,7 @@
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.read.database.CountDatabasePlan;
import org.apache.iotdb.confignode.consensus.request.read.database.GetDatabasePlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetAllSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetAllTemplateSetInfoPlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetPathsSetTemplatePlan;
@@ -62,6 +63,7 @@
import org.apache.iotdb.confignode.consensus.response.database.CountDatabaseResp;
import org.apache.iotdb.confignode.consensus.response.database.DatabaseSchemaResp;
import org.apache.iotdb.confignode.consensus.response.partition.PathInfoResp;
+import org.apache.iotdb.confignode.consensus.response.table.ShowTableResp;
import org.apache.iotdb.confignode.consensus.response.template.AllTemplateSetInfoResp;
import org.apache.iotdb.confignode.consensus.response.template.TemplateInfoResp;
import org.apache.iotdb.confignode.consensus.response.template.TemplateSetInfoResp;
@@ -1060,7 +1062,15 @@
// region table management
public TShowTableResp showTables(final String database) {
- return clusterSchemaInfo.showTables(database);
+ try {
+ return ((ShowTableResp) configManager.getConsensusManager().read(new ShowTablePlan(database)))
+ .convertToTShowTableResp();
+ } catch (final ConsensusException e) {
+ LOGGER.warn("Failed in the read API executing the consensus layer due to: ", e);
+ final TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ res.setMessage(e.getMessage());
+ return new TShowTableResp(res);
+ }
}
public byte[] getAllTableInfoForDataNodeActivation() {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
index 1b3a648..a931962 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
@@ -44,6 +44,7 @@
import org.apache.iotdb.confignode.consensus.request.read.pipe.plugin.GetPipePluginJarPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionIdPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.CheckTemplateSettablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetPathsSetTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetSchemaTemplatePlan;
@@ -306,6 +307,8 @@
return clusterSchemaInfo.getAllTemplateSetInfo();
case GetTemplateSetInfo:
return clusterSchemaInfo.getTemplateSetInfo((GetTemplateSetInfoPlan) req);
+ case ShowTable:
+ return clusterSchemaInfo.showTables((ShowTablePlan) req);
case GetTriggerTable:
return triggerInfo.getTriggerTable((GetTriggerTablePlan) req);
case GetTriggerLocation:
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
index d4139ff..982c145 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
@@ -35,6 +35,7 @@
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.confignode.consensus.request.read.database.CountDatabasePlan;
import org.apache.iotdb.confignode.consensus.request.read.database.GetDatabasePlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.CheckTemplateSettablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetPathsSetTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetSchemaTemplatePlan;
@@ -62,12 +63,12 @@
import org.apache.iotdb.confignode.consensus.response.database.CountDatabaseResp;
import org.apache.iotdb.confignode.consensus.response.database.DatabaseSchemaResp;
import org.apache.iotdb.confignode.consensus.response.partition.PathInfoResp;
+import org.apache.iotdb.confignode.consensus.response.table.ShowTableResp;
import org.apache.iotdb.confignode.consensus.response.template.AllTemplateSetInfoResp;
import org.apache.iotdb.confignode.consensus.response.template.TemplateInfoResp;
import org.apache.iotdb.confignode.consensus.response.template.TemplateSetInfoResp;
import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
-import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TTableInfo;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
import org.apache.iotdb.db.schemaengine.template.Template;
@@ -1076,24 +1077,26 @@
}
}
- public TShowTableResp showTables(final String database) {
+ public ShowTableResp showTables(final ShowTablePlan plan) {
databaseReadWriteLock.readLock().lock();
try {
- return new TShowTableResp(StatusUtils.OK)
- .setTableInfoList(
- mTree
- .getAllUsingTablesUnderSpecificDatabase(getQualifiedDatabasePartialPath(database))
- .stream()
- .map(
- tsTable ->
- new TTableInfo(
- tsTable.getTableName(),
- tsTable
- .getPropValue(TTL_PROPERTY.toLowerCase(Locale.ENGLISH))
- .orElse(TTL_INFINITE)))
- .collect(Collectors.toList()));
+ return new ShowTableResp(
+ StatusUtils.OK,
+ mTree
+ .getAllUsingTablesUnderSpecificDatabase(
+ getQualifiedDatabasePartialPath(plan.getDatabase()))
+ .stream()
+ .map(
+ tsTable ->
+ new TTableInfo(
+ tsTable.getTableName(),
+ tsTable
+ .getPropValue(TTL_PROPERTY.toLowerCase(Locale.ENGLISH))
+ .orElse(TTL_INFINITE)))
+ .collect(Collectors.toList()));
} catch (final MetadataException e) {
- return new TShowTableResp(RpcUtils.getStatus(e.getErrorCode(), e.getMessage()));
+ return new ShowTableResp(
+ RpcUtils.getStatus(e.getErrorCode(), e.getMessage()), Collections.emptyList());
} finally {
databaseReadWriteLock.readLock().unlock();
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index 86fde27..f0ea3ff 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -78,6 +78,7 @@
import org.apache.iotdb.confignode.consensus.request.read.pipe.task.ShowPipePlanV2;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionIdPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetAllSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetAllTemplateSetInfoPlan;
import org.apache.iotdb.confignode.consensus.request.read.template.GetPathsSetTemplatePlan;
@@ -939,17 +940,17 @@
@Test
public void GetRegionLocationsPlanTest() throws IOException {
- GetRegionInfoListPlan req0 = new GetRegionInfoListPlan();
- TShowRegionReq showRegionReq = new TShowRegionReq();
+ final GetRegionInfoListPlan req0 = new GetRegionInfoListPlan();
+ final TShowRegionReq showRegionReq = new TShowRegionReq();
req0.setShowRegionReq(showRegionReq);
showRegionReq.setConsensusGroupType(TConsensusGroupType.DataRegion);
- GetRegionInfoListPlan req1 =
+ final GetRegionInfoListPlan req1 =
(GetRegionInfoListPlan) ConfigPhysicalPlan.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0.getType(), req1.getType());
Assert.assertEquals(req0.getShowRegionReq(), req1.getShowRegionReq());
final List<String> sgList = Collections.singletonList("root.sg1, root.sg2, root.*");
showRegionReq.setDatabases(new ArrayList<>(sgList));
- GetRegionInfoListPlan req2 =
+ final GetRegionInfoListPlan req2 =
(GetRegionInfoListPlan) ConfigPhysicalPlan.Factory.create(req0.serializeToByteBuffer());
Assert.assertEquals(req0.getType(), req1.getType());
Assert.assertEquals(req0.getShowRegionReq(), req2.getShowRegionReq());
@@ -957,20 +958,21 @@
@Test
public void CreateSchemaTemplatePlanTest() throws IOException, IllegalPathException {
- Template template = newSchemaTemplate("template_name");
- CreateSchemaTemplatePlan createSchemaTemplatePlan0 =
+ final Template template = newSchemaTemplate("template_name");
+ final CreateSchemaTemplatePlan createSchemaTemplatePlan0 =
new CreateSchemaTemplatePlan(template.serialize().array());
- CreateSchemaTemplatePlan createSchemaTemplatePlan1 =
+ final CreateSchemaTemplatePlan createSchemaTemplatePlan1 =
(CreateSchemaTemplatePlan)
ConfigPhysicalPlan.Factory.create(createSchemaTemplatePlan0.serializeToByteBuffer());
Assert.assertEquals(createSchemaTemplatePlan0, createSchemaTemplatePlan1);
}
private Template newSchemaTemplate(String name) throws IllegalPathException {
- List<String> measurements = Arrays.asList(name + "_" + "temperature", name + "_" + "status");
- List<TSDataType> dataTypes = Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN);
- List<TSEncoding> encodings = Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN);
- List<CompressionType> compressors =
+ final List<String> measurements =
+ Arrays.asList(name + "_" + "temperature", name + "_" + "status");
+ final List<TSDataType> dataTypes = Arrays.asList(TSDataType.FLOAT, TSDataType.BOOLEAN);
+ final List<TSEncoding> encodings = Arrays.asList(TSEncoding.RLE, TSEncoding.PLAIN);
+ final List<CompressionType> compressors =
Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY);
return new Template(name, measurements, dataTypes, encodings, compressors);
}
@@ -991,8 +993,8 @@
@Test
public void GetSchemaTemplatePlanTest() throws IOException {
- GetSchemaTemplatePlan getSchemaTemplatePlan = new GetSchemaTemplatePlan("template1");
- GetSchemaTemplatePlan deserializedPlan =
+ final GetSchemaTemplatePlan getSchemaTemplatePlan = new GetSchemaTemplatePlan("template1");
+ final GetSchemaTemplatePlan deserializedPlan =
(GetSchemaTemplatePlan)
ConfigPhysicalPlan.Factory.create(getSchemaTemplatePlan.serializeToByteBuffer());
Assert.assertEquals("template1", deserializedPlan.getTemplateName());
@@ -1000,7 +1002,7 @@
@Test
public void GetAllSchemaTemplatePlanTest() throws IOException {
- GetAllSchemaTemplatePlan getAllSchemaTemplatePlan0 = new GetAllSchemaTemplatePlan();
+ final GetAllSchemaTemplatePlan getAllSchemaTemplatePlan0 = new GetAllSchemaTemplatePlan();
Assert.assertTrue(
ConfigPhysicalPlan.Factory.create(getAllSchemaTemplatePlan0.serializeToByteBuffer())
instanceof GetAllSchemaTemplatePlan);
@@ -1008,8 +1010,9 @@
@Test
public void GetNodesInSchemaTemplatePlanTest() throws IOException {
- GetSchemaTemplatePlan getSchemaTemplatePlan0 = new GetSchemaTemplatePlan("template_name_test");
- GetSchemaTemplatePlan getSchemaTemplatePlan1 =
+ final GetSchemaTemplatePlan getSchemaTemplatePlan0 =
+ new GetSchemaTemplatePlan("template_name_test");
+ final GetSchemaTemplatePlan getSchemaTemplatePlan1 =
(GetSchemaTemplatePlan)
ConfigPhysicalPlan.Factory.create(getSchemaTemplatePlan0.serializeToByteBuffer());
Assert.assertEquals(getSchemaTemplatePlan0, getSchemaTemplatePlan1);
@@ -1017,10 +1020,10 @@
@Test
public void GetNodePathsPartitionPlanTest() throws IOException, IllegalPathException {
- GetNodePathsPartitionPlan getNodePathsPartitionPlan0 = new GetNodePathsPartitionPlan();
+ final GetNodePathsPartitionPlan getNodePathsPartitionPlan0 = new GetNodePathsPartitionPlan();
getNodePathsPartitionPlan0.setPartialPath(new PartialPath("root.sg1.**"));
getNodePathsPartitionPlan0.setScope(ALL_MATCH_SCOPE);
- GetNodePathsPartitionPlan getNodePathsPartitionPlan1 =
+ final GetNodePathsPartitionPlan getNodePathsPartitionPlan1 =
(GetNodePathsPartitionPlan)
ConfigPhysicalPlan.Factory.create(getNodePathsPartitionPlan0.serializeToByteBuffer());
Assert.assertEquals(getNodePathsPartitionPlan0, getNodePathsPartitionPlan1);
@@ -1028,7 +1031,7 @@
@Test
public void GetAllTemplateSetInfoPlanTest() throws IOException {
- GetAllTemplateSetInfoPlan getAllTemplateSetInfoPlan = new GetAllTemplateSetInfoPlan();
+ final GetAllTemplateSetInfoPlan getAllTemplateSetInfoPlan = new GetAllTemplateSetInfoPlan();
Assert.assertTrue(
ConfigPhysicalPlan.Factory.create(getAllTemplateSetInfoPlan.serializeToByteBuffer())
instanceof GetAllTemplateSetInfoPlan);
@@ -1036,9 +1039,9 @@
@Test
public void SetSchemaTemplatePlanTest() throws IOException {
- SetSchemaTemplatePlan setSchemaTemplatePlanPlan0 =
+ final SetSchemaTemplatePlan setSchemaTemplatePlanPlan0 =
new SetSchemaTemplatePlan("template_name_test", "root.in.sg.dw");
- SetSchemaTemplatePlan setSchemaTemplatePlanPlan1 =
+ final SetSchemaTemplatePlan setSchemaTemplatePlanPlan1 =
(SetSchemaTemplatePlan)
ConfigPhysicalPlan.Factory.create(setSchemaTemplatePlanPlan0.serializeToByteBuffer());
Assert.assertEquals(
@@ -1048,9 +1051,9 @@
@Test
public void ShowPathSetTemplatePlanTest() throws IOException {
- GetPathsSetTemplatePlan getPathsSetTemplatePlan0 =
+ final GetPathsSetTemplatePlan getPathsSetTemplatePlan0 =
new GetPathsSetTemplatePlan("template_name_test", ALL_MATCH_SCOPE);
- GetPathsSetTemplatePlan getPathsSetTemplatePlan1 =
+ final GetPathsSetTemplatePlan getPathsSetTemplatePlan1 =
(GetPathsSetTemplatePlan)
ConfigPhysicalPlan.Factory.create(getPathsSetTemplatePlan0.serializeToByteBuffer());
Assert.assertEquals(getPathsSetTemplatePlan0.getName(), getPathsSetTemplatePlan1.getName());
@@ -1058,8 +1061,8 @@
@Test
public void DropSchemaTemplateTest() throws IOException {
- DropSchemaTemplatePlan dropSchemaTemplatePlan = new DropSchemaTemplatePlan("template");
- DropSchemaTemplatePlan deserializedPlan =
+ final DropSchemaTemplatePlan dropSchemaTemplatePlan = new DropSchemaTemplatePlan("template");
+ final DropSchemaTemplatePlan deserializedPlan =
(DropSchemaTemplatePlan)
ConfigPhysicalPlan.Factory.create(dropSchemaTemplatePlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1068,16 +1071,16 @@
@Test
public void CreatePipeSinkPlanTest() throws IOException {
- Map<String, String> attributes = new HashMap<>();
+ final Map<String, String> attributes = new HashMap<>();
attributes.put("ip", "127.0.0.1");
attributes.put("port", "6667");
- TPipeSinkInfo pipeSinkInfo =
+ final TPipeSinkInfo pipeSinkInfo =
new TPipeSinkInfo()
.setPipeSinkName("demo")
.setPipeSinkType("IoTDB")
.setAttributes(attributes);
- CreatePipeSinkPlanV1 createPipeSinkPlan = new CreatePipeSinkPlanV1(pipeSinkInfo);
- CreatePipeSinkPlanV1 createPipeSinkPlan1 =
+ final CreatePipeSinkPlanV1 createPipeSinkPlan = new CreatePipeSinkPlanV1(pipeSinkInfo);
+ final CreatePipeSinkPlanV1 createPipeSinkPlan1 =
(CreatePipeSinkPlanV1)
ConfigPhysicalPlan.Factory.create(createPipeSinkPlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1086,8 +1089,8 @@
@Test
public void DropPipeSinkPlanTest() throws IOException {
- DropPipeSinkPlanV1 dropPipeSinkPlan = new DropPipeSinkPlanV1("demo");
- DropPipeSinkPlanV1 dropPipeSinkPlan1 =
+ final DropPipeSinkPlanV1 dropPipeSinkPlan = new DropPipeSinkPlanV1("demo");
+ final DropPipeSinkPlanV1 dropPipeSinkPlan1 =
(DropPipeSinkPlanV1)
ConfigPhysicalPlan.Factory.create(dropPipeSinkPlan.serializeToByteBuffer());
Assert.assertEquals(dropPipeSinkPlan.getPipeSinkName(), dropPipeSinkPlan1.getPipeSinkName());
@@ -1095,13 +1098,13 @@
@Test
public void GetPipeSinkPlanTest() throws IOException {
- GetPipeSinkPlanV1 getPipeSinkPlan = new GetPipeSinkPlanV1("demo");
- GetPipeSinkPlanV1 getPipeSinkPlan1 =
+ final GetPipeSinkPlanV1 getPipeSinkPlan = new GetPipeSinkPlanV1("demo");
+ final GetPipeSinkPlanV1 getPipeSinkPlan1 =
(GetPipeSinkPlanV1)
ConfigPhysicalPlan.Factory.create(getPipeSinkPlan.serializeToByteBuffer());
Assert.assertEquals(getPipeSinkPlan.getPipeSinkName(), getPipeSinkPlan1.getPipeSinkName());
- GetPipeSinkPlanV1 getPipeSinkPlanWithNullName = new GetPipeSinkPlanV1();
- GetPipeSinkPlanV1 getPipeSinkPlanWithNullName1 =
+ final GetPipeSinkPlanV1 getPipeSinkPlanWithNullName = new GetPipeSinkPlanV1();
+ final GetPipeSinkPlanV1 getPipeSinkPlanWithNullName1 =
(GetPipeSinkPlanV1)
ConfigPhysicalPlan.Factory.create(getPipeSinkPlanWithNullName.serializeToByteBuffer());
Assert.assertEquals(
@@ -1111,11 +1114,11 @@
@Test
public void PreCreatePipePlanTest() throws IOException {
- PipeInfo pipeInfo =
+ final PipeInfo pipeInfo =
new TsFilePipeInfo(
"name", "demo", PipeStatus.PARTIAL_CREATE, System.currentTimeMillis(), 999, false);
- PreCreatePipePlanV1 PreCreatePipePlan = new PreCreatePipePlanV1(pipeInfo);
- PreCreatePipePlanV1 PreCreatePipePlan1 =
+ final PreCreatePipePlanV1 PreCreatePipePlan = new PreCreatePipePlanV1(pipeInfo);
+ final PreCreatePipePlanV1 PreCreatePipePlan1 =
(PreCreatePipePlanV1)
ConfigPhysicalPlan.Factory.create(PreCreatePipePlan.serializeToByteBuffer());
Assert.assertEquals(PreCreatePipePlan.getPipeInfo(), PreCreatePipePlan1.getPipeInfo());
@@ -1123,10 +1126,10 @@
@Test
public void RecordPipeMessagePlanTest() throws IOException {
- RecordPipeMessagePlan recordPipeMessagePlan =
+ final RecordPipeMessagePlan recordPipeMessagePlan =
new RecordPipeMessagePlan(
"testPipe", new PipeMessage(PipeMessage.PipeMessageType.ERROR, "testError"));
- RecordPipeMessagePlan recordPipeMessagePlan1 =
+ final RecordPipeMessagePlan recordPipeMessagePlan1 =
(RecordPipeMessagePlan)
ConfigPhysicalPlan.Factory.create(recordPipeMessagePlan.serializeToByteBuffer());
Assert.assertEquals(recordPipeMessagePlan.getPipeName(), recordPipeMessagePlan1.getPipeName());
@@ -1140,9 +1143,9 @@
@Test
public void SetPipeStatusPlanTest() throws IOException {
- SetPipeStatusPlanV1 setPipeStatusPlan =
+ final SetPipeStatusPlanV1 setPipeStatusPlan =
new SetPipeStatusPlanV1("pipe", PipeStatus.PARTIAL_CREATE);
- SetPipeStatusPlanV1 setPipeStatusPlan1 =
+ final SetPipeStatusPlanV1 setPipeStatusPlan1 =
(SetPipeStatusPlanV1)
ConfigPhysicalPlan.Factory.create(setPipeStatusPlan.serializeToByteBuffer());
Assert.assertEquals(setPipeStatusPlan.getPipeName(), setPipeStatusPlan1.getPipeName());
@@ -1151,21 +1154,21 @@
@Test
public void CreatePipePlanV2Test() throws IOException {
- Map<String, String> extractorAttributes = new HashMap<>();
- Map<String, String> processorAttributes = new HashMap<>();
- Map<String, String> connectorAttributes = new HashMap<>();
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
extractorAttributes.put("extractor", "org.apache.iotdb.pipe.extractor.DefaultExtractor");
processorAttributes.put("processor", "org.apache.iotdb.pipe.processor.SDTFilterProcessor");
connectorAttributes.put("connector", "org.apache.iotdb.pipe.protocol.ThriftTransporter");
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
- PipeStaticMeta pipeStaticMeta =
+ final PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
"testPipe", 121, extractorAttributes, processorAttributes, connectorAttributes);
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
- CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta, pipeRuntimeMeta);
- CreatePipePlanV2 createPipePlanV21 =
+ final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ final CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta, pipeRuntimeMeta);
+ final CreatePipePlanV2 createPipePlanV21 =
(CreatePipePlanV2)
ConfigPhysicalPlan.Factory.create(createPipePlanV2.serializeToByteBuffer());
Assert.assertEquals(
@@ -1176,21 +1179,21 @@
@Test
public void AlterPipePlanV2Test() throws IOException {
- Map<String, String> extractorAttributes = new HashMap<>();
- Map<String, String> processorAttributes = new HashMap<>();
- Map<String, String> connectorAttributes = new HashMap<>();
+ final Map<String, String> extractorAttributes = new HashMap<>();
+ final Map<String, String> processorAttributes = new HashMap<>();
+ final Map<String, String> connectorAttributes = new HashMap<>();
extractorAttributes.put("pattern", "root.db");
processorAttributes.put("processor", "do-nothing-processor");
connectorAttributes.put("batch.enable", "false");
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
+ final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
- PipeStaticMeta pipeStaticMeta =
+ final PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
"testPipe", 121, extractorAttributes, processorAttributes, connectorAttributes);
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
- AlterPipePlanV2 alterPipePlanV2 = new AlterPipePlanV2(pipeStaticMeta, pipeRuntimeMeta);
- AlterPipePlanV2 alterPipePlanV21 =
+ final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ final AlterPipePlanV2 alterPipePlanV2 = new AlterPipePlanV2(pipeStaticMeta, pipeRuntimeMeta);
+ final AlterPipePlanV2 alterPipePlanV21 =
(AlterPipePlanV2)
ConfigPhysicalPlan.Factory.create(alterPipePlanV2.serializeToByteBuffer());
Assert.assertEquals(alterPipePlanV2.getPipeStaticMeta(), alterPipePlanV21.getPipeStaticMeta());
@@ -1200,9 +1203,9 @@
@Test
public void SetPipeStatusPlanV2Test() throws IOException {
- SetPipeStatusPlanV2 setPipeStatusPlanV2 =
+ final SetPipeStatusPlanV2 setPipeStatusPlanV2 =
new SetPipeStatusPlanV2("pipe", org.apache.iotdb.commons.pipe.task.meta.PipeStatus.RUNNING);
- SetPipeStatusPlanV2 setPipeStatusPlanV21 =
+ final SetPipeStatusPlanV2 setPipeStatusPlanV21 =
(SetPipeStatusPlanV2)
ConfigPhysicalPlan.Factory.create(setPipeStatusPlanV2.serializeToByteBuffer());
Assert.assertEquals(setPipeStatusPlanV2.getPipeName(), setPipeStatusPlanV21.getPipeName());
@@ -1211,55 +1214,55 @@
@Test
public void DropPipePlanV2Test() throws IOException {
- DropPipePlanV2 dropPipePlanV2 = new DropPipePlanV2("demo");
- DropPipePlanV2 dropPipePlanV21 =
+ final DropPipePlanV2 dropPipePlanV2 = new DropPipePlanV2("demo");
+ final DropPipePlanV2 dropPipePlanV21 =
(DropPipePlanV2) ConfigPhysicalPlan.Factory.create(dropPipePlanV2.serializeToByteBuffer());
Assert.assertEquals(dropPipePlanV2.getPipeName(), dropPipePlanV21.getPipeName());
}
@Test
public void OperateMultiplePipesPlanV2Test() throws IOException {
- PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
- ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
+ final PipeTaskMeta pipeTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 1);
+ final ConcurrentMap<Integer, PipeTaskMeta> pipeTasks = new ConcurrentHashMap<>();
pipeTasks.put(1, pipeTaskMeta);
- PipeStaticMeta pipeStaticMeta =
+ final PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
"testCreate",
5,
Collections.singletonMap("k1", "v1"),
Collections.singletonMap("k2", "v2"),
Collections.singletonMap("k3", "v3"));
- PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
- CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta, pipeRuntimeMeta);
+ final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta(pipeTasks);
+ final CreatePipePlanV2 createPipePlanV2 = new CreatePipePlanV2(pipeStaticMeta, pipeRuntimeMeta);
- PipeTaskMeta pipeTaskMeta1 = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
- ConcurrentMap<Integer, PipeTaskMeta> pipeTasks1 = new ConcurrentHashMap<>();
+ final PipeTaskMeta pipeTaskMeta1 = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 2);
+ final ConcurrentMap<Integer, PipeTaskMeta> pipeTasks1 = new ConcurrentHashMap<>();
pipeTasks.put(2, pipeTaskMeta1);
- PipeStaticMeta pipeStaticMeta1 =
+ final PipeStaticMeta pipeStaticMeta1 =
new PipeStaticMeta(
"testAlter",
6,
Collections.singletonMap("k4", "v4"),
Collections.singletonMap("k5", "v5"),
Collections.singletonMap("k6", "v6"));
- PipeRuntimeMeta pipeRuntimeMeta1 = new PipeRuntimeMeta(pipeTasks1);
- AlterPipePlanV2 alterPipePlanV2 = new AlterPipePlanV2(pipeStaticMeta1, pipeRuntimeMeta1);
+ final PipeRuntimeMeta pipeRuntimeMeta1 = new PipeRuntimeMeta(pipeTasks1);
+ final AlterPipePlanV2 alterPipePlanV2 = new AlterPipePlanV2(pipeStaticMeta1, pipeRuntimeMeta1);
- DropPipePlanV2 dropPipePlanV2 = new DropPipePlanV2("testDrop");
+ final DropPipePlanV2 dropPipePlanV2 = new DropPipePlanV2("testDrop");
- SetPipeStatusPlanV2 setPipeStatusPlanV2 =
+ final SetPipeStatusPlanV2 setPipeStatusPlanV2 =
new SetPipeStatusPlanV2(
"testSet", org.apache.iotdb.commons.pipe.task.meta.PipeStatus.RUNNING);
- List<ConfigPhysicalPlan> subPlans = new ArrayList<>();
+ final List<ConfigPhysicalPlan> subPlans = new ArrayList<>();
subPlans.add(createPipePlanV2);
subPlans.add(alterPipePlanV2);
subPlans.add(dropPipePlanV2);
subPlans.add(setPipeStatusPlanV2);
- OperateMultiplePipesPlanV2 operateMultiplePipesPlanV2 =
+ final OperateMultiplePipesPlanV2 operateMultiplePipesPlanV2 =
new OperateMultiplePipesPlanV2(subPlans);
- OperateMultiplePipesPlanV2 operateMultiplePipesPlanV21 =
+ final OperateMultiplePipesPlanV2 operateMultiplePipesPlanV21 =
(OperateMultiplePipesPlanV2)
ConfigPhysicalPlan.Factory.create(operateMultiplePipesPlanV2.serializeToByteBuffer());
Assert.assertEquals(
@@ -1267,13 +1270,13 @@
}
@Test
- public void ShowPipePlanTest() throws IOException {
- ShowPipePlanV1 showPipePlan = new ShowPipePlanV1("demo");
- ShowPipePlanV1 showPipePlan1 =
+ public void ShowPipePlanV1Test() throws IOException {
+ final ShowPipePlanV1 showPipePlan = new ShowPipePlanV1("demo");
+ final ShowPipePlanV1 showPipePlan1 =
(ShowPipePlanV1) ConfigPhysicalPlan.Factory.create(showPipePlan.serializeToByteBuffer());
Assert.assertEquals(showPipePlan.getPipeName(), showPipePlan1.getPipeName());
- ShowPipePlanV1 showPipePlanWithNullName = new ShowPipePlanV1();
- ShowPipePlanV1 showPipePlanWithNullName1 =
+ final ShowPipePlanV1 showPipePlanWithNullName = new ShowPipePlanV1();
+ final ShowPipePlanV1 showPipePlanWithNullName1 =
(ShowPipePlanV1)
ConfigPhysicalPlan.Factory.create(showPipePlanWithNullName.serializeToByteBuffer());
Assert.assertEquals(
@@ -1282,11 +1285,11 @@
@Test
public void CreatePipePluginPlanTest() throws IOException {
- CreatePipePluginPlan createPipePluginPlan =
+ final CreatePipePluginPlan createPipePluginPlan =
new CreatePipePluginPlan(
new PipePluginMeta("testPlugin", "org.apache.iotdb.TestJar", false, "test.jar", "???"),
new Binary("123", TSFileConfig.STRING_CHARSET));
- CreatePipePluginPlan createPipePluginPlan1 =
+ final CreatePipePluginPlan createPipePluginPlan1 =
(CreatePipePluginPlan)
ConfigPhysicalPlan.Factory.create(createPipePluginPlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1296,8 +1299,8 @@
@Test
public void DropPipePluginPlanTest() throws IOException {
- DropPipePluginPlan dropPipePluginPlan = new DropPipePluginPlan("testPlugin");
- DropPipePluginPlan dropPipePluginPlan1 =
+ final DropPipePluginPlan dropPipePluginPlan = new DropPipePluginPlan("testPlugin");
+ final DropPipePluginPlan dropPipePluginPlan1 =
(DropPipePluginPlan)
ConfigPhysicalPlan.Factory.create(dropPipePluginPlan.serializeToByteBuffer());
Assert.assertEquals(dropPipePluginPlan.getPluginName(), dropPipePluginPlan1.getPluginName());
@@ -1305,16 +1308,16 @@
@Test
public void pipeHandleLeaderChangePlanTest() throws IOException {
- Map<TConsensusGroupId, Integer> newLeaderMap = new HashMap<>();
+ final Map<TConsensusGroupId, Integer> newLeaderMap = new HashMap<>();
// Do not test SchemaRegion or ConfigRegion since the Type is always "DataRegion" when
// deserialized
newLeaderMap.put(new TConsensusGroupId(TConsensusGroupType.DataRegion, 1), 2);
newLeaderMap.put(new TConsensusGroupId(TConsensusGroupType.DataRegion, 2), 3);
newLeaderMap.put(new TConsensusGroupId(TConsensusGroupType.DataRegion, 3), 5);
- PipeHandleLeaderChangePlan pipeHandleLeaderChangePlan =
+ final PipeHandleLeaderChangePlan pipeHandleLeaderChangePlan =
new PipeHandleLeaderChangePlan(newLeaderMap);
- PipeHandleLeaderChangePlan pipeHandleLeaderChangePlan1 =
+ final PipeHandleLeaderChangePlan pipeHandleLeaderChangePlan1 =
(PipeHandleLeaderChangePlan)
ConfigPhysicalPlan.Factory.create(pipeHandleLeaderChangePlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1324,8 +1327,8 @@
@Test
public void pipeHandleMetaChangePlanTest() throws IOException {
- List<PipeMeta> pipeMetaList = new ArrayList<>();
- PipeStaticMeta pipeStaticMeta =
+ final List<PipeMeta> pipeMetaList = new ArrayList<>();
+ final PipeStaticMeta pipeStaticMeta =
new PipeStaticMeta(
"pipeName",
123L,
@@ -1341,7 +1344,7 @@
}
},
new HashMap<String, String>() {});
- PipeRuntimeMeta pipeRuntimeMeta =
+ final PipeRuntimeMeta pipeRuntimeMeta =
new PipeRuntimeMeta(
new ConcurrentHashMap<Integer, PipeTaskMeta>() {
{
@@ -1350,8 +1353,9 @@
}
});
pipeMetaList.add(new PipeMeta(pipeStaticMeta, pipeRuntimeMeta));
- PipeHandleMetaChangePlan pipeHandleMetaChangePlan1 = new PipeHandleMetaChangePlan(pipeMetaList);
- PipeHandleMetaChangePlan pipeHandleMetaChangePlan2 =
+ final PipeHandleMetaChangePlan pipeHandleMetaChangePlan1 =
+ new PipeHandleMetaChangePlan(pipeMetaList);
+ final PipeHandleMetaChangePlan pipeHandleMetaChangePlan2 =
(PipeHandleMetaChangePlan)
ConfigPhysicalPlan.Factory.create(pipeHandleMetaChangePlan1.serializeToByteBuffer());
Assert.assertEquals(
@@ -1360,12 +1364,12 @@
@Test
public void CreateTopicPlanTest() throws IOException {
- Map<String, String> attributes = new HashMap<>();
+ final Map<String, String> attributes = new HashMap<>();
attributes.put("k1", "v1");
attributes.put("k2", "v2");
- CreateTopicPlan createTopicPlan =
+ final CreateTopicPlan createTopicPlan =
new CreateTopicPlan(new TopicMeta("test_topic", 1, attributes));
- CreateTopicPlan createTopicPlan1 =
+ final CreateTopicPlan createTopicPlan1 =
(CreateTopicPlan)
ConfigPhysicalPlan.Factory.create(createTopicPlan.serializeToByteBuffer());
Assert.assertEquals(createTopicPlan.getTopicMeta(), createTopicPlan1.getTopicMeta());
@@ -1373,32 +1377,33 @@
@Test
public void DropTopicPlanTest() throws IOException {
- DropTopicPlan dropTopicPlan = new DropTopicPlan("test_topic");
- DropTopicPlan dropTopicPlan1 =
+ final DropTopicPlan dropTopicPlan = new DropTopicPlan("test_topic");
+ final DropTopicPlan dropTopicPlan1 =
(DropTopicPlan) ConfigPhysicalPlan.Factory.create(dropTopicPlan.serializeToByteBuffer());
Assert.assertEquals(dropTopicPlan.getTopicName(), dropTopicPlan1.getTopicName());
}
@Test
public void AlterTopicPlanTest() throws IOException {
- Map<String, String> attributes = new HashMap<>();
+ final Map<String, String> attributes = new HashMap<>();
attributes.put("k1", "v1");
attributes.put("k2", "v2");
- AlterTopicPlan alterTopicPlan = new AlterTopicPlan(new TopicMeta("test_topic", 1, attributes));
- AlterTopicPlan alterTopicPlan1 =
+ final AlterTopicPlan alterTopicPlan =
+ new AlterTopicPlan(new TopicMeta("test_topic", 1, attributes));
+ final AlterTopicPlan alterTopicPlan1 =
(AlterTopicPlan) ConfigPhysicalPlan.Factory.create(alterTopicPlan.serializeToByteBuffer());
Assert.assertEquals(alterTopicPlan.getTopicMeta(), alterTopicPlan1.getTopicMeta());
}
@Test
public void AlterMultipleTopicsTopicPlanTest() throws IOException {
- List<AlterTopicPlan> subPlans = new ArrayList<>();
+ final List<AlterTopicPlan> subPlans = new ArrayList<>();
subPlans.add(
new AlterTopicPlan(new TopicMeta("test_topic1", 1, Collections.singletonMap("k1", "v1"))));
subPlans.add(
new AlterTopicPlan(new TopicMeta("test_topic2", 2, Collections.singletonMap("k2", "v2"))));
- AlterMultipleTopicsPlan alterMultipleTopicsPlan = new AlterMultipleTopicsPlan(subPlans);
- AlterMultipleTopicsPlan alterMultipleTopicsPlan1 =
+ final AlterMultipleTopicsPlan alterMultipleTopicsPlan = new AlterMultipleTopicsPlan(subPlans);
+ final AlterMultipleTopicsPlan alterMultipleTopicsPlan1 =
(AlterMultipleTopicsPlan)
ConfigPhysicalPlan.Factory.create(alterMultipleTopicsPlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1407,11 +1412,12 @@
@Test
public void TopicHandleMetaChangePlanTest() throws IOException {
- List<TopicMeta> topicMetas = new ArrayList<>();
+ final List<TopicMeta> topicMetas = new ArrayList<>();
topicMetas.add(new TopicMeta("topic1", 1, Collections.singletonMap("k1", "v1")));
topicMetas.add(new TopicMeta("topic2", 2, Collections.singletonMap("k2", "v2")));
- TopicHandleMetaChangePlan topicHandleMetaChangePlan = new TopicHandleMetaChangePlan(topicMetas);
- TopicHandleMetaChangePlan topicHandleMetaChangePlan1 =
+ final TopicHandleMetaChangePlan topicHandleMetaChangePlan =
+ new TopicHandleMetaChangePlan(topicMetas);
+ final TopicHandleMetaChangePlan topicHandleMetaChangePlan1 =
(TopicHandleMetaChangePlan)
ConfigPhysicalPlan.Factory.create(topicHandleMetaChangePlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1421,14 +1427,14 @@
@Test
public void AlterConsumerGroupPlanTest() throws IOException {
- Map<String, String> attributes = new HashMap<>();
+ final Map<String, String> attributes = new HashMap<>();
attributes.put("k1", "v1");
attributes.put("k2", "v2");
- AlterConsumerGroupPlan alterConsumerGroupPlan =
+ final AlterConsumerGroupPlan alterConsumerGroupPlan =
new AlterConsumerGroupPlan(
new ConsumerGroupMeta(
"test_consumer_group", 1, new ConsumerMeta("test_consumer", 2, attributes)));
- AlterConsumerGroupPlan alterConsumerGroupPlan1 =
+ final AlterConsumerGroupPlan alterConsumerGroupPlan1 =
(AlterConsumerGroupPlan)
ConfigPhysicalPlan.Factory.create(alterConsumerGroupPlan.serializeToByteBuffer());
Assert.assertEquals(
@@ -1438,16 +1444,16 @@
@Test
public void ConsumerGroupHandleMetaChangePlanTest() throws IOException {
- List<ConsumerGroupMeta> consumerGroupMetas = new ArrayList<>();
+ final List<ConsumerGroupMeta> consumerGroupMetas = new ArrayList<>();
consumerGroupMetas.add(
new ConsumerGroupMeta(
"cg1", 1, new ConsumerMeta("c1", 11, Collections.singletonMap("k1", "v1"))));
consumerGroupMetas.add(
new ConsumerGroupMeta(
"cg2", 2, new ConsumerMeta("c2", 22, Collections.singletonMap("k2", "v2"))));
- ConsumerGroupHandleMetaChangePlan consumerGroupHandleMetaChangePlan =
+ final ConsumerGroupHandleMetaChangePlan consumerGroupHandleMetaChangePlan =
new ConsumerGroupHandleMetaChangePlan(consumerGroupMetas);
- ConsumerGroupHandleMetaChangePlan consumerGroupHandleMetaChangePlan1 =
+ final ConsumerGroupHandleMetaChangePlan consumerGroupHandleMetaChangePlan1 =
(ConsumerGroupHandleMetaChangePlan)
ConfigPhysicalPlan.Factory.create(
consumerGroupHandleMetaChangePlan.serializeToByteBuffer());
@@ -1540,6 +1546,15 @@
}
@Test
+ public void showTablePlanTest() throws IOException {
+ final ShowTablePlan showTablePlan = new ShowTablePlan("database");
+ Assert.assertEquals(
+ showTablePlan.getDatabase(),
+ ((ShowTablePlan) ConfigPhysicalPlan.Factory.create(showTablePlan.serializeToByteBuffer()))
+ .getDatabase());
+ }
+
+ @Test
public void GetTriggerTablePlanTest() throws IOException {
GetTriggerTablePlan getTriggerTablePlan0 = new GetTriggerTablePlan(true);
GetTriggerTablePlan getTriggerTablePlan1 =
@@ -1680,12 +1695,12 @@
@Test
public void GetTriggerJarPlanTest() throws IOException {
- List<String> jarNames = new ArrayList<>();
+ final List<String> jarNames = new ArrayList<>();
jarNames.add("test1");
jarNames.add("test2");
- GetTriggerJarPlan getTriggerJarPlan0 = new GetTriggerJarPlan(jarNames);
+ final GetTriggerJarPlan getTriggerJarPlan0 = new GetTriggerJarPlan(jarNames);
- GetTriggerJarPlan getTriggerJarPlan1 =
+ final GetTriggerJarPlan getTriggerJarPlan1 =
(GetTriggerJarPlan)
ConfigPhysicalPlan.Factory.create(getTriggerJarPlan0.serializeToByteBuffer());
Assert.assertEquals(getTriggerJarPlan0.getJarNames(), getTriggerJarPlan1.getJarNames());
@@ -1693,8 +1708,8 @@
@Test
public void GetRegionIdPlanTest() throws IOException {
- GetRegionIdPlan getRegionIdPlan0 = new GetRegionIdPlan(ConfigRegion);
- GetRegionIdPlan getRegionIdPlan1 =
+ final GetRegionIdPlan getRegionIdPlan0 = new GetRegionIdPlan(ConfigRegion);
+ final GetRegionIdPlan getRegionIdPlan1 =
(GetRegionIdPlan)
ConfigPhysicalPlan.Factory.create(getRegionIdPlan0.serializeToByteBuffer());
Assert.assertEquals(getRegionIdPlan0, getRegionIdPlan1);
@@ -1702,8 +1717,8 @@
@Test
public void GetTimeSlotListPlanTest() throws IOException {
- GetTimeSlotListPlan getTimeSlotListPlan0 = new GetTimeSlotListPlan(0, Long.MAX_VALUE);
- GetTimeSlotListPlan getTimeSlotListPlan1 =
+ final GetTimeSlotListPlan getTimeSlotListPlan0 = new GetTimeSlotListPlan(0, Long.MAX_VALUE);
+ final GetTimeSlotListPlan getTimeSlotListPlan1 =
(GetTimeSlotListPlan)
ConfigPhysicalPlan.Factory.create(getTimeSlotListPlan0.serializeToByteBuffer());
Assert.assertEquals(getTimeSlotListPlan0, getTimeSlotListPlan1);
@@ -1711,8 +1726,9 @@
@Test
public void CountTimeSlotListPlanTest() throws IOException {
- CountTimeSlotListPlan countTimeSlotListPlan0 = new CountTimeSlotListPlan(0, Long.MAX_VALUE);
- CountTimeSlotListPlan countTimeSlotListPlan1 =
+ final CountTimeSlotListPlan countTimeSlotListPlan0 =
+ new CountTimeSlotListPlan(0, Long.MAX_VALUE);
+ final CountTimeSlotListPlan countTimeSlotListPlan1 =
(CountTimeSlotListPlan)
ConfigPhysicalPlan.Factory.create(countTimeSlotListPlan0.serializeToByteBuffer());
Assert.assertEquals(countTimeSlotListPlan0, countTimeSlotListPlan1);
@@ -1720,9 +1736,9 @@
@Test
public void GetSeriesSlotListPlanTest() throws IOException {
- GetSeriesSlotListPlan getSeriesSlotListPlan0 =
+ final GetSeriesSlotListPlan getSeriesSlotListPlan0 =
new GetSeriesSlotListPlan("root.test", SchemaRegion);
- GetSeriesSlotListPlan getSeriesSlotListPlan1 =
+ final GetSeriesSlotListPlan getSeriesSlotListPlan1 =
(GetSeriesSlotListPlan)
ConfigPhysicalPlan.Factory.create(getSeriesSlotListPlan0.serializeToByteBuffer());
Assert.assertEquals(getSeriesSlotListPlan0, getSeriesSlotListPlan1);
@@ -1730,11 +1746,11 @@
@Test
public void GetPipePluginJarPlanTest() throws IOException {
- List<String> jarNames = new ArrayList<>();
+ final List<String> jarNames = new ArrayList<>();
jarNames.add("org.apache.testJar");
jarNames.add("org.apache.testJar2");
- GetPipePluginJarPlan getPipePluginJarPlan0 = new GetPipePluginJarPlan(jarNames);
- GetPipePluginJarPlan getPipePluginJarPlan1 =
+ final GetPipePluginJarPlan getPipePluginJarPlan0 = new GetPipePluginJarPlan(jarNames);
+ final GetPipePluginJarPlan getPipePluginJarPlan1 =
(GetPipePluginJarPlan)
ConfigPhysicalPlan.Factory.create(getPipePluginJarPlan0.serializeToByteBuffer());
Assert.assertEquals(getPipePluginJarPlan0, getPipePluginJarPlan1);
@@ -1742,8 +1758,8 @@
@Test
public void GetPipePluginTablePlanTest() throws IOException {
- GetPipePluginTablePlan getPipePluginTablePlan0 = new GetPipePluginTablePlan();
- GetPipePluginTablePlan getPipePluginTablePlan1 =
+ final GetPipePluginTablePlan getPipePluginTablePlan0 = new GetPipePluginTablePlan();
+ final GetPipePluginTablePlan getPipePluginTablePlan1 =
(GetPipePluginTablePlan)
ConfigPhysicalPlan.Factory.create(getPipePluginTablePlan0.serializeToByteBuffer());
Assert.assertEquals(getPipePluginTablePlan0, getPipePluginTablePlan1);
@@ -1751,8 +1767,8 @@
@Test
public void ShowPipePlanV2Test() throws IOException {
- ShowPipePlanV2 showPipePlanV2 = new ShowPipePlanV2();
- ShowPipePlanV2 showPipePlanV21 =
+ final ShowPipePlanV2 showPipePlanV2 = new ShowPipePlanV2();
+ final ShowPipePlanV2 showPipePlanV21 =
(ShowPipePlanV2) ConfigPhysicalPlan.Factory.create(showPipePlanV2.serializeToByteBuffer());
Assert.assertEquals(showPipePlanV2, showPipePlanV21);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
index 0ab7fc3..2e94559 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -34,6 +34,8 @@
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT;
import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
@@ -85,13 +87,17 @@
private void saveUpdatedTableInfo(
final Map<String, List<TsTable>> tableMap,
final Map<String, Map<String, TsTable>> localTableMap) {
- for (final Map.Entry<String, List<TsTable>> entry : tableMap.entrySet()) {
- final Map<String, TsTable> map = new ConcurrentHashMap<>();
- for (final TsTable table : entry.getValue()) {
- map.put(table.getTableName(), table);
- }
- localTableMap.put(entry.getKey(), map);
- }
+ tableMap.forEach(
+ (key, value) ->
+ localTableMap.put(
+ key,
+ value.stream()
+ .collect(
+ Collectors.toMap(
+ TsTable::getTableName,
+ Function.identity(),
+ (v1, v2) -> v2,
+ ConcurrentHashMap::new))));
}
@Override