Use consensus read in show table

Caideyipi · Sep 18, 2024 · ebe3768 · ebe3768
1 parent 43710fd
commit ebe3768
Show file tree

Hide file tree

Showing 14 changed files with 339 additions and 203 deletions.
diff --git a/...gnode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/...gnode/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 @@ public static ConfigPhysicalPlan create(ByteBuffer buffer) throws IOException {
         case SetTableProperties:
           plan = new SetTablePropertiesPlan();
           break;
+        case ShowTable:
+          plan = new ShowTablePlan();
+          break;
         case GetNodePathsPartition:
           plan = new GetNodePathsPartitionPlan();
           break;

diff --git a/...e/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/...e/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
@@ -160,6 +160,7 @@ public enum ConfigPhysicalPlanType {
   CommitCreateTable((short) 852),
   AddTableColumn((short) 853),
   SetTableProperties((short) 854),
+  ShowTable((short) 855),
 
   /** Deprecated types for sync, restored them for upgrade. */
   @Deprecated

diff --git a/.../org/apache/iotdb/confignode/consensus/request/read/pipe/plugin/GetPipePluginJarPlan.java b/.../org/apache/iotdb/confignode/consensus/request/read/pipe/plugin/GetPipePluginJarPlan.java
@@ -37,7 +37,7 @@ public GetPipePluginJarPlan() {
     super(ConfigPhysicalPlanType.GetPipePluginJar);
   }
 
-  public GetPipePluginJarPlan(List<String> jarNames) {
+  public GetPipePluginJarPlan(final List<String> jarNames) {
     super(ConfigPhysicalPlanType.GetPipePluginJar);
     this.jarNames = jarNames;
   }
@@ -47,18 +47,18 @@ public List<String> getJarNames() {
   }
 
   @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/...ain/java/org/apache/iotdb/confignode/consensus/request/read/pipe/task/ShowPipePlanV2.java b/...ain/java/org/apache/iotdb/confignode/consensus/request/read/pipe/task/ShowPipePlanV2.java
@@ -33,12 +33,12 @@ public ShowPipePlanV2() {
   }
 
   @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/...src/main/java/org/apache/iotdb/confignode/consensus/request/read/table/ShowTablePlan.java b/...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/...src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java b/...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 List<PipeMeta> getAllPipeMeta() {
   }
 
   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);
-
-        return new PipeTableResp(
-            status,
-            allPipeMeta.stream()
-                .filter(
-                    pipeMeta ->
-                        pipeMeta
-                            .getStaticMeta()
-                            .getConnectorParameters()
-                            .toString()
-                            .equals(sortedConnectorParametersString))
-                .collect(Collectors.toList()));
-      }
+      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()));
     }
   }
 

diff --git a/...ode/src/main/java/org/apache/iotdb/confignode/consensus/response/table/ShowTableResp.java b/...ode/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
@@ -2576,11 +2576,9 @@ public TSStatus alterTable(final TAlterTableReq req) {
   @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/...n/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java b/...n/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java
@@ -181,14 +181,14 @@ public TSStatus dropPipe(TDropPipeReq req) {
                 "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/...ignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/...ignode/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 @@ public synchronized TSStatus extendSchemaTemplate(
   // 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/...de/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/...de/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 @@ public DataSet executeQueryPlan(ConfigPhysicalPlan req)
         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/...gnode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java b/...gnode/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 TSStatus commitCreateTable(final CommitCreateTablePlan plan) {
     }
   }
 
-  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();
     }