You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2024/04/14 10:46:08 UTC
(iotdb) branch ty/TableModelGrammar updated: [Table Model] Support basic table management
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ty/TableModelGrammar
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/ty/TableModelGrammar by this push:
new 6408bd86367 [Table Model] Support basic table management
6408bd86367 is described below
commit 6408bd8636787964624c9c36296408d411cd54ff
Author: Marcos_Zyk <38...@users.noreply.github.com>
AuthorDate: Sun Apr 14 18:46:02 2024 +0800
[Table Model] Support basic table management
---
.../java/org/apache/iotdb/rpc/TSStatusCode.java | 3 +
.../confignode/client/DataNodeRequestType.java | 3 +
.../client/async/AsyncDataNodeClientPool.java | 7 +
.../client/async/handlers/AsyncClientHandler.java | 1 +
.../client/sync/SyncDataNodeClientPool.java | 3 +
.../consensus/request/ConfigPhysicalPlan.java | 12 +
.../consensus/request/ConfigPhysicalPlanType.java | 5 +
.../request/write/table/CommitCreateTablePlan.java | 66 ++++
.../request/write/table/PreCreateTablePlan.java | 67 ++++
.../write/table/RollbackCreateTablePlan.java | 66 ++++
.../iotdb/confignode/manager/ConfigManager.java | 13 +
.../iotdb/confignode/manager/ProcedureManager.java | 45 +++
.../iotdb/confignode/manager/node/NodeManager.java | 2 +
.../manager/schema/ClusterSchemaManager.java | 10 +
.../persistence/executor/ConfigPlanExecutor.java | 9 +
.../persistence/schema/ClusterSchemaInfo.java | 75 ++++
.../confignode/persistence/schema/ConfigMTree.java | 109 ++++++
.../schema/mnode/impl/ConfigTableNode.java | 221 +++++++++++
.../schema/mnode/impl/TableNodeStatus.java | 58 +++
.../schema/mnode/info/ConfigTableInfo.java} | 40 +-
.../impl/schema/DataNodeRegionTaskExecutor.java | 2 +-
.../impl/schema/table/CreateTableProcedure.java | 429 +++++++++++++++++++++
.../procedure/state/schema/CreateTableState.java | 28 ++
.../procedure/store/ProcedureFactory.java | 6 +
.../confignode/procedure/store/ProcedureType.java | 2 +
.../thrift/ConfigNodeRPCServiceProcessor.java | 5 +
.../table/TableAlreadyExistsException.java | 32 ++
.../metadata/table/TableNotExistsException.java | 32 ++
.../iotdb/db/protocol/client/ConfigNodeClient.java | 7 +
.../impl/DataNodeInternalRPCServiceImpl.java | 39 ++
.../common/header/ColumnHeaderConstant.java | 17 +
.../common/header/DatasetHeaderFactory.java | 8 +
.../iotdb/db/queryengine/plan/Coordinator.java | 2 +-
.../queryengine/plan/analyze/AnalyzeVisitor.java | 18 +
.../plan/analyze/lock/SchemaLockType.java | 17 +-
.../execution/config/TableConfigTaskVisitor.java | 109 +++++-
.../config/executor/ClusterConfigTaskExecutor.java | 69 +++-
.../config/executor/IConfigTaskExecutor.java | 8 +-
.../metadata/relational/CreateTableTask.java | 16 +-
.../metadata/relational/DescribeTableTask.java | 85 ++++
.../config/metadata/relational/ShowTablesTask.java | 77 ++++
.../db/schemaengine/table/DataNodeTableCache.java | 207 ++++++++++
.../iotdb/db/schemaengine/table/ITableCache.java | 33 ++
.../java/org/apache/iotdb/db/service/DataNode.java | 4 +
.../iotdb/commons/schema/SchemaConstant.java | 6 +
.../apache/iotdb/commons/schema/table/TsTable.java | 144 +++++++
.../schema/table/TsTableInternalRPCType.java | 64 +++
.../schema/table/TsTableInternalRPCUtil.java | 98 +++++
.../schema/table/column/AttributeColumnSchema.java | 57 +++
.../schema/table/column/IdColumnSchema.java | 57 +++
.../table/column/MeasurementColumnSchema.java | 101 +++++
.../schema/table/column/TimeColumnSchema.java | 57 +++
.../schema/table/column/TsTableColumnCategory.java | 73 ++++
.../schema/table/column/TsTableColumnSchema.java | 74 ++++
.../table/column/TsTableColumnSchemaUtil.java | 80 ++++
.../src/main/thrift/confignode.thrift | 8 +
.../src/main/thrift/datanode.thrift | 10 +
57 files changed, 2863 insertions(+), 33 deletions(-)
diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
index 5e7fcb543c6..50724c932fb 100644
--- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
+++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java
@@ -84,6 +84,9 @@ public enum TSStatusCode {
MEASUREMENT_ALREADY_EXISTS_IN_TEMPLATE(527),
TYPE_NOT_FOUND(528),
+ TABLE_NOT_EXISTS(550),
+ TABLE_ALREADY_EXISTS(551),
+
// Storage Engine
SYSTEM_READ_ONLY(600),
STORAGE_ENGINE_ERROR(601),
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
index 1c7bfa27530..a597f5fccbe 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
@@ -115,4 +115,7 @@ public enum DataNodeRequestType {
// Quota
SET_SPACE_QUOTA,
SET_THROTTLE_QUOTA,
+
+ // Table
+ UPDATE_TABLE,
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
index e52367276c8..4fb4b7a7277 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
@@ -78,6 +78,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListWithTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackViewSchemaBlackListReq;
+import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTriggerLocationReq;
@@ -491,6 +492,12 @@ public class AsyncDataNodeClientPool {
(AsyncTSStatusRPCHandler)
clientHandler.createAsyncRPCHandler(requestId, targetDataNode));
break;
+ case UPDATE_TABLE:
+ client.updateTable(
+ (TUpdateTableReq) clientHandler.getRequest(requestId),
+ (AsyncTSStatusRPCHandler)
+ clientHandler.createAsyncRPCHandler(requestId, targetDataNode));
+ break;
default:
LOGGER.error(
"Unexpected DataNode Request Type: {} when sendAsyncRequestToDataNode",
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
index a8c445665d0..c5dc81e1868 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
@@ -264,6 +264,7 @@ public class AsyncClientHandler<Q, R> {
case UPDATE_REGION_ROUTE_MAP:
case INVALIDATE_MATCHED_SCHEMA_CACHE:
case UPDATE_TEMPLATE:
+ case UPDATE_TABLE:
case CHANGE_REGION_LEADER:
case KILL_QUERY_INSTANCE:
case RESET_PEER_LIST:
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
index 32d09ca5191..6d3c266e764 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
+import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -127,6 +128,8 @@ public class SyncDataNodeClientPool {
return client.killQueryInstance((String) req);
case UPDATE_TEMPLATE:
return client.updateTemplate((TUpdateTemplateReq) req);
+ case UPDATE_TABLE:
+ return client.updateTable((TUpdateTableReq) req);
case CREATE_NEW_REGION_PEER:
return client.createNewRegionPeer((TCreatePeerReq) req);
case ADD_REGION_PEER:
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 17a08c1a2d9..cf8dc435b56 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
@@ -115,6 +115,9 @@ import org.apache.iotdb.confignode.consensus.request.write.sync.PreCreatePipePla
import org.apache.iotdb.confignode.consensus.request.write.sync.RecordPipeMessagePlan;
import org.apache.iotdb.confignode.consensus.request.write.sync.SetPipeStatusPlanV1;
import org.apache.iotdb.confignode.consensus.request.write.sync.ShowPipePlanV1;
+import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.DropSchemaTemplatePlan;
@@ -395,6 +398,15 @@ public abstract class ConfigPhysicalPlan implements IConsensusRequest {
case ExtendSchemaTemplate:
plan = new ExtendSchemaTemplatePlan();
break;
+ case PreCreateTable:
+ plan = new PreCreateTablePlan();
+ break;
+ case RollbackCreateTable:
+ plan = new RollbackCreateTablePlan();
+ break;
+ case CommitCreateTable:
+ plan = new CommitCreateTablePlan();
+ 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 4e547304eb9..a5f7b5c35d6 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
@@ -146,6 +146,11 @@ public enum ConfigPhysicalPlanType {
CommitSetSchemaTemplate((short) 813),
ExtendSchemaTemplate((short) 814),
+ /* Table */
+ PreCreateTable((short) 850),
+ RollbackCreateTable((short) 851),
+ CommitCreateTable((short) 852),
+
/** Deprecated types for sync, restored them for upgrade. */
@Deprecated
CreatePipeSinkV1((short) 900),
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/CommitCreateTablePlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/CommitCreateTablePlan.java
new file mode 100644
index 00000000000..3637c2d29f4
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/CommitCreateTablePlan.java
@@ -0,0 +1,66 @@
+/*
+ * 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.write.table;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class CommitCreateTablePlan extends ConfigPhysicalPlan {
+
+ private String database;
+
+ private String tableName;
+
+ public CommitCreateTablePlan() {
+ super(ConfigPhysicalPlanType.CommitCreateTable);
+ }
+
+ public CommitCreateTablePlan(String database, String tableName) {
+ super(ConfigPhysicalPlanType.CommitCreateTable);
+ this.database = database;
+ this.tableName = tableName;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ReadWriteIOUtils.write(database, stream);
+ ReadWriteIOUtils.write(tableName, stream);
+ }
+
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ this.database = ReadWriteIOUtils.readString(buffer);
+ this.tableName = ReadWriteIOUtils.readString(buffer);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/PreCreateTablePlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/PreCreateTablePlan.java
new file mode 100644
index 00000000000..b0baa451bff
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/PreCreateTablePlan.java
@@ -0,0 +1,67 @@
+/*
+ * 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.write.table;
+
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class PreCreateTablePlan extends ConfigPhysicalPlan {
+
+ private String database;
+
+ private TsTable table;
+
+ public PreCreateTablePlan() {
+ super(ConfigPhysicalPlanType.PreCreateTable);
+ }
+
+ public PreCreateTablePlan(String database, TsTable table) {
+ super(ConfigPhysicalPlanType.PreCreateTable);
+ this.database = database;
+ this.table = table;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public TsTable getTable() {
+ return table;
+ }
+
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ReadWriteIOUtils.write(database, stream);
+ table.serialize(stream);
+ }
+
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ database = ReadWriteIOUtils.readString(buffer);
+ table = TsTable.deserialize(buffer);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackCreateTablePlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackCreateTablePlan.java
new file mode 100644
index 00000000000..9dfe994a23a
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackCreateTablePlan.java
@@ -0,0 +1,66 @@
+/*
+ * 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.write.table;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class RollbackCreateTablePlan extends ConfigPhysicalPlan {
+
+ private String database;
+
+ private String tableName;
+
+ public RollbackCreateTablePlan() {
+ super(ConfigPhysicalPlanType.RollbackCreateTable);
+ }
+
+ public RollbackCreateTablePlan(String database, String tableName) {
+ super(ConfigPhysicalPlanType.RollbackCreateTable);
+ this.database = database;
+ this.tableName = tableName;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ @Override
+ protected void serializeImpl(DataOutputStream stream) throws IOException {
+ stream.writeShort(getType().getPlanType());
+ ReadWriteIOUtils.write(database, stream);
+ ReadWriteIOUtils.write(tableName, stream);
+ }
+
+ @Override
+ protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+ this.database = ReadWriteIOUtils.readString(buffer);
+ this.tableName = ReadWriteIOUtils.readString(buffer);
+ }
+}
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 e25d26d7e8a..31f85c6ca1c 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
@@ -43,6 +43,8 @@ import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.path.PathPatternUtil;
import org.apache.iotdb.commons.pipe.connector.payload.airgap.AirGapPseudoTPipeTransferRequest;
import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.utils.AuthUtils;
import org.apache.iotdb.commons.utils.PathUtils;
@@ -2235,4 +2237,15 @@ public class ConfigManager implements IManager {
? clusterQuotaManager.getThrottleQuota()
: new TThrottleQuotaResp(status);
}
+
+ public TSStatus createTable(ByteBuffer tableInfo) {
+ TSStatus status = confirmLeader();
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ Pair<String, TsTable> pair =
+ TsTableInternalRPCUtil.deserializeSingleTsTable(tableInfo.array());
+ return procedureManager.createTable(pair.left, pair.right);
+ } else {
+ return status;
+ }
+ }
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index f5898d4bc17..c91ffebceb6 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -33,6 +33,7 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathDeserializeUtil;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.pipe.plugin.meta.PipePluginMeta;
+import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.trigger.TriggerInformation;
@@ -76,6 +77,7 @@ import org.apache.iotdb.confignode.procedure.impl.schema.DeleteLogicalViewProced
import org.apache.iotdb.confignode.procedure.impl.schema.DeleteTimeSeriesProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.SetTemplateProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.UnsetTemplateProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.table.CreateTableProcedure;
import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.CreateConsumerProcedure;
import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.DropConsumerProcedure;
import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.runtime.ConsumerGroupMetaSyncProcedure;
@@ -1203,6 +1205,49 @@ public class ProcedureManager {
}
}
+ public TSStatus createTable(String database, TsTable table) {
+ long procedureId = -1;
+ synchronized (this) {
+ boolean hasOverlappedTask = false;
+ ProcedureType type;
+ CreateTableProcedure createTableProcedure;
+ for (Procedure<?> procedure : executor.getProcedures().values()) {
+ type = ProcedureFactory.getProcedureType(procedure);
+ if (type == null || !type.equals(ProcedureType.CREATE_TABLE_PROCEDURE)) {
+ continue;
+ }
+ createTableProcedure = (CreateTableProcedure) procedure;
+ if (database.equals(createTableProcedure.getDatabase())
+ && table.equals(createTableProcedure.getTable())) {
+ procedureId = createTableProcedure.getProcId();
+ break;
+ }
+ if (database.equals(createTableProcedure.getDatabase())
+ && table.getTableName().equals(createTableProcedure.getTable().getTableName())) {
+ hasOverlappedTask = true;
+ break;
+ }
+ }
+
+ if (procedureId == -1) {
+ if (hasOverlappedTask) {
+ return RpcUtils.getStatus(
+ TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
+ "Some other task is creating table with same name.");
+ }
+ procedureId = this.executor.submitProcedure(new CreateTableProcedure(database, table));
+ }
+ }
+ List<TSStatus> procedureStatus = new ArrayList<>();
+ boolean isSucceed =
+ waitingProcedureFinished(Collections.singletonList(procedureId), procedureStatus);
+ if (isSucceed) {
+ return StatusUtils.OK;
+ } else {
+ return procedureStatus.get(0);
+ }
+ }
+
// ======================================================
/*
GET-SET Region
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index f66663a9483..560f4e7a9b5 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -243,6 +243,8 @@ public class NodeManager {
getPipeManager().getPipePluginCoordinator().getPipePluginTable().getAllPipePluginMeta());
runtimeConfiguration.setAllTTLInformation(
DataNodeRegisterResp.convertAllTTLInformation(getClusterSchemaManager().getAllTTLInfo()));
+ runtimeConfiguration.setTableInfo(
+ getClusterSchemaManager().getAllTableInfoForDataNodeActivation());
return runtimeConfiguration;
} finally {
getTriggerManager().getTriggerInfo().releaseTriggerTableLock();
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 8d26049ce62..034ba3bdaea 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
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.commons.utils.StatusUtils;
@@ -1129,6 +1130,15 @@ public class ClusterSchemaManager {
}
}
+ // region table management
+
+ public byte[] getAllTableInfoForDataNodeActivation() {
+ return TsTableInternalRPCUtil.serializeTableInitializationInfo(
+ clusterSchemaInfo.getAllUsingTables(), clusterSchemaInfo.getAllPreCreateTables());
+ }
+
+ // endregion
+
/**
* Only leader use this interface. Get the remain schema quota of specified schema region.
*
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 99b4fe3d31e..f4fb2442bb1 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
@@ -97,6 +97,9 @@ import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.Al
import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan;
import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.DropTopicPlan;
import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.runtime.TopicHandleMetaChangePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.DropSchemaTemplatePlan;
@@ -443,6 +446,12 @@ public class ConfigPlanExecutor {
return clusterSchemaInfo.dropSchemaTemplate((DropSchemaTemplatePlan) physicalPlan);
case ExtendSchemaTemplate:
return clusterSchemaInfo.extendSchemaTemplate((ExtendSchemaTemplatePlan) physicalPlan);
+ case PreCreateTable:
+ return clusterSchemaInfo.preCreateTable((PreCreateTablePlan) physicalPlan);
+ case RollbackCreateTable:
+ return clusterSchemaInfo.rollbackCreateTable((RollbackCreateTablePlan) physicalPlan);
+ case CommitCreateTable:
+ return clusterSchemaInfo.commitCreateTable((CommitCreateTablePlan) physicalPlan);
case CreatePipeV2:
return pipeInfo.createPipe((CreatePipePlanV2) physicalPlan);
case SetPipeStatusV2:
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 57102d73aa4..2140a74aa3c 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
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.commons.utils.TestOnly;
@@ -44,6 +45,9 @@ import org.apache.iotdb.confignode.consensus.request.write.database.SetDataRepli
import org.apache.iotdb.confignode.consensus.request.write.database.SetSchemaReplicationFactorPlan;
import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
import org.apache.iotdb.confignode.consensus.request.write.database.SetTimePartitionIntervalPlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CommitSetSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.CreateSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.DropSchemaTemplatePlan;
@@ -92,6 +96,7 @@ import static org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCAR
import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN;
import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_SCOPE;
import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_TEMPLATE;
+import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
import static org.apache.iotdb.commons.schema.SchemaConstant.SYSTEM_DATABASE_PATTERN;
/**
@@ -1039,6 +1044,76 @@ public class ClusterSchemaInfo implements SnapshotProcessor {
return schemaMap;
}
+ // region table management
+
+ public TSStatus preCreateTable(PreCreateTablePlan preCreateTablePlan) {
+ databaseReadWriteLock.writeLock().lock();
+ try {
+ mTree.preCreateTable(
+ new PartialPath(new String[] {ROOT, preCreateTablePlan.getDatabase()}),
+ preCreateTablePlan.getTable());
+ return RpcUtils.SUCCESS_STATUS;
+ } catch (MetadataException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } finally {
+ databaseReadWriteLock.writeLock().unlock();
+ }
+ }
+
+ public TSStatus rollbackCreateTable(RollbackCreateTablePlan rollbackCreateTablePlan) {
+ databaseReadWriteLock.writeLock().lock();
+ try {
+ mTree.rollbackCreateTable(
+ new PartialPath(new String[] {ROOT, rollbackCreateTablePlan.getDatabase()}),
+ rollbackCreateTablePlan.getTableName());
+ return RpcUtils.SUCCESS_STATUS;
+ } catch (MetadataException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } finally {
+ databaseReadWriteLock.writeLock().unlock();
+ }
+ }
+
+ public TSStatus commitCreateTable(CommitCreateTablePlan commitCreateTablePlan) {
+ databaseReadWriteLock.writeLock().lock();
+ try {
+ mTree.commitCreateTable(
+ new PartialPath(new String[] {ROOT, commitCreateTablePlan.getDatabase()}),
+ commitCreateTablePlan.getTableName());
+ return RpcUtils.SUCCESS_STATUS;
+ } catch (MetadataException e) {
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } finally {
+ databaseReadWriteLock.writeLock().unlock();
+ }
+ }
+
+ public Map<String, List<TsTable>> getAllUsingTables() {
+ databaseReadWriteLock.readLock().lock();
+ try {
+ return mTree.getAllUsingTables();
+ } catch (MetadataException e) {
+ LOGGER.warn(e.getMessage(), e);
+ throw new RuntimeException(e);
+ } finally {
+ databaseReadWriteLock.readLock().unlock();
+ }
+ }
+
+ public Map<String, List<TsTable>> getAllPreCreateTables() {
+ databaseReadWriteLock.readLock().lock();
+ try {
+ return mTree.getAllPreCreateTables();
+ } catch (MetadataException e) {
+ LOGGER.warn(e.getMessage(), e);
+ throw new RuntimeException(e);
+ } finally {
+ databaseReadWriteLock.readLock().unlock();
+ }
+ }
+
+ // endregion
+
@TestOnly
public void clear() {
mTree.clear();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
index 9f4a226daca..c44c697d0a6 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
@@ -28,12 +28,18 @@ import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode;
import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
import org.apache.iotdb.commons.schema.node.utils.IMNodeIterator;
+import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.commons.utils.ThriftConfigNodeSerDeUtils;
import org.apache.iotdb.confignode.persistence.schema.mnode.IConfigMNode;
import org.apache.iotdb.confignode.persistence.schema.mnode.factory.ConfigMNodeFactory;
+import org.apache.iotdb.confignode.persistence.schema.mnode.impl.ConfigTableNode;
+import org.apache.iotdb.confignode.persistence.schema.mnode.impl.TableNodeStatus;
import org.apache.iotdb.db.exception.metadata.DatabaseAlreadySetException;
import org.apache.iotdb.db.exception.metadata.DatabaseNotSetException;
+import org.apache.iotdb.db.exception.metadata.PathAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
+import org.apache.iotdb.db.exception.metadata.table.TableAlreadyExistsException;
+import org.apache.iotdb.db.exception.metadata.table.TableNotExistsException;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.DatabaseCollector;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.MNodeAboveDBCollector;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.traverser.collector.MNodeCollector;
@@ -67,7 +73,9 @@ import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_RESULT_NODES;
import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_TEMPLATE;
import static org.apache.iotdb.commons.schema.SchemaConstant.INTERNAL_MNODE_TYPE;
import static org.apache.iotdb.commons.schema.SchemaConstant.NON_TEMPLATE;
+import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
import static org.apache.iotdb.commons.schema.SchemaConstant.STORAGE_GROUP_MNODE_TYPE;
+import static org.apache.iotdb.commons.schema.SchemaConstant.TABLE_MNODE_TYPE;
// Since the ConfigMTree is all stored in memory, thus it is not restricted to manage MNode through
// MTreeStore.
@@ -621,6 +629,84 @@ public class ConfigMTree {
// endregion
+ // region table management
+
+ public void preCreateTable(PartialPath database, TsTable table) throws MetadataException {
+ IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode();
+ IConfigMNode node = databaseNode.getChild(table.getTableName());
+ if (node == null) {
+ ConfigTableNode tableNode =
+ (ConfigTableNode)
+ databaseNode.addChild(
+ table.getTableName(), new ConfigTableNode(databaseNode, table.getTableName()));
+ tableNode.setTable(table);
+ tableNode.setStatus(TableNodeStatus.PRE_CREATE);
+ } else if (node instanceof ConfigTableNode) {
+ throw new TableAlreadyExistsException(
+ database.getFullPath().substring(ROOT.length() + 1), table.getTableName());
+ } else {
+ throw new PathAlreadyExistException(database.concatNode(table.getTableName()).getFullPath());
+ }
+ }
+
+ public void rollbackCreateTable(PartialPath database, String tableName) throws MetadataException {
+ IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode();
+ databaseNode.deleteChild(tableName);
+ }
+
+ public void commitCreateTable(PartialPath database, String tableName) throws MetadataException {
+ IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(database).getAsMNode();
+ if (!databaseNode.hasChild(tableName)) {
+ throw new TableNotExistsException(
+ database.getFullPath().substring(ROOT.length() + 1), tableName);
+ }
+ ConfigTableNode tableNode = (ConfigTableNode) databaseNode.getChild(tableName);
+ if (!tableNode.getStatus().equals(TableNodeStatus.PRE_CREATE)) {
+ throw new IllegalStateException();
+ }
+ tableNode.setStatus(TableNodeStatus.USING);
+ }
+
+ public Map<String, List<TsTable>> getAllUsingTables() throws MetadataException {
+ Map<String, List<TsTable>> result = new HashMap<>();
+ List<PartialPath> databaseList = getAllDatabasePaths();
+ for (PartialPath databasePath : databaseList) {
+ String database = databasePath.getFullPath().substring(ROOT.length() + 1);
+ IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(databasePath).getAsMNode();
+ for (IConfigMNode child : databaseNode.getChildren().values()) {
+ if (child instanceof ConfigTableNode) {
+ ConfigTableNode tableNode = (ConfigTableNode) child;
+ if (!tableNode.getStatus().equals(TableNodeStatus.USING)) {
+ continue;
+ }
+ result.computeIfAbsent(database, k -> new ArrayList<>()).add(tableNode.getTable());
+ }
+ }
+ }
+ return result;
+ }
+
+ public Map<String, List<TsTable>> getAllPreCreateTables() throws MetadataException {
+ Map<String, List<TsTable>> result = new HashMap<>();
+ List<PartialPath> databaseList = getAllDatabasePaths();
+ for (PartialPath databasePath : databaseList) {
+ String database = databasePath.getFullPath().substring(ROOT.length() + 1);
+ IConfigMNode databaseNode = getDatabaseNodeByDatabasePath(databasePath).getAsMNode();
+ for (IConfigMNode child : databaseNode.getChildren().values()) {
+ if (child instanceof ConfigTableNode) {
+ ConfigTableNode tableNode = (ConfigTableNode) child;
+ if (!tableNode.getStatus().equals(TableNodeStatus.PRE_CREATE)) {
+ continue;
+ }
+ result.computeIfAbsent(database, k -> new ArrayList<>()).add(tableNode.getTable());
+ }
+ }
+ }
+ return result;
+ }
+
+ // endregion
+
// region Serialization and Deserialization
public void serialize(OutputStream outputStream) throws IOException {
@@ -641,6 +727,8 @@ public class ConfigMTree {
for (IConfigMNode child : node.getChildren().values()) {
if (child.isDatabase()) {
serializeDatabaseNode(child.getAsDatabaseMNode(), outputStream);
+ } else if (child instanceof ConfigTableNode) {
+ serializeTableNode((ConfigTableNode) node, outputStream);
} else {
serializeConfigBasicMNode(child, outputStream);
}
@@ -658,6 +746,14 @@ public class ConfigMTree {
storageGroupNode.getAsMNode().getDatabaseSchema(), outputStream);
}
+ private void serializeTableNode(ConfigTableNode tableNode, OutputStream outputStream)
+ throws IOException {
+ ReadWriteIOUtils.write(TABLE_MNODE_TYPE, outputStream);
+ ReadWriteIOUtils.write(tableNode.getName(), outputStream);
+ tableNode.getTable().serialize(outputStream);
+ tableNode.getStatus().serialize(outputStream);
+ }
+
public void deserialize(InputStream inputStream) throws IOException {
byte type = ReadWriteIOUtils.readByte(inputStream);
@@ -666,6 +762,7 @@ public class ConfigMTree {
Stack<Pair<IConfigMNode, Boolean>> stack = new Stack<>();
IConfigMNode databaseMNode;
IConfigMNode internalMNode;
+ IConfigMNode tableNode;
if (type == STORAGE_GROUP_MNODE_TYPE) {
databaseMNode = deserializeDatabaseMNode(inputStream);
@@ -702,6 +799,11 @@ public class ConfigMTree {
stack.push(new Pair<>(databaseMNode, true));
name = databaseMNode.getName();
break;
+ case TABLE_MNODE_TYPE:
+ tableNode = deserializeTableMNode(inputStream).getAsMNode();
+ stack.push(new Pair<>(tableNode, false));
+ name = tableNode.getName();
+ break;
default:
logger.error("Unrecognized node type. Cannot deserialize MTreeAboveSG from given buffer");
return;
@@ -727,5 +829,12 @@ public class ConfigMTree {
return databaseMNode.getAsMNode();
}
+ private IConfigMNode deserializeTableMNode(InputStream inputStream) throws IOException {
+ ConfigTableNode tableNode = new ConfigTableNode(null, ReadWriteIOUtils.readString(inputStream));
+ tableNode.setTable(TsTable.deserialize(inputStream));
+ tableNode.setStatus(TableNodeStatus.deserialize(inputStream));
+ return tableNode;
+ }
+
// endregion
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/impl/ConfigTableNode.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/impl/ConfigTableNode.java
new file mode 100644
index 00000000000..41fb3bfe469
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/impl/ConfigTableNode.java
@@ -0,0 +1,221 @@
+/*
+ * 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.persistence.schema.mnode.impl;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.node.MNodeType;
+import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode;
+import org.apache.iotdb.commons.schema.node.utils.IMNodeContainer;
+import org.apache.iotdb.commons.schema.node.visitor.MNodeVisitor;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.confignode.persistence.schema.mnode.IConfigMNode;
+import org.apache.iotdb.confignode.persistence.schema.mnode.container.ConfigMNodeContainer;
+import org.apache.iotdb.confignode.persistence.schema.mnode.info.ConfigTableInfo;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ConfigTableNode implements IConfigMNode {
+
+ private IConfigMNode parent;
+
+ private transient String fullPath;
+
+ private ConfigTableInfo tableNodeInfo;
+
+ public ConfigTableNode(IConfigMNode parent, String name) {
+ this.parent = parent;
+ this.tableNodeInfo = new ConfigTableInfo(name);
+ }
+
+ public TsTable getTable() {
+ return tableNodeInfo.getTable();
+ }
+
+ public void setTable(TsTable table) {
+ tableNodeInfo.setTable(table);
+ }
+
+ public TableNodeStatus getStatus() {
+ return tableNodeInfo.getStatus();
+ }
+
+ public void setStatus(TableNodeStatus status) {
+ tableNodeInfo.setStatus(status);
+ }
+
+ @Override
+ public String getName() {
+ return tableNodeInfo.getName();
+ }
+
+ @Override
+ public void setName(String name) {
+ tableNodeInfo.setName(name);
+ }
+
+ @Override
+ public IConfigMNode getParent() {
+ return parent;
+ }
+
+ @Override
+ public void setParent(IConfigMNode parent) {
+ this.parent = parent;
+ }
+
+ @Override
+ public String getFullPath() {
+ if (fullPath == null) {
+ fullPath = concatFullPath();
+ }
+ return fullPath;
+ }
+
+ String concatFullPath() {
+ StringBuilder builder = new StringBuilder(getName());
+ IConfigMNode curr = this;
+ while (curr.getParent() != null) {
+ curr = curr.getParent();
+ builder.insert(0, IoTDBConstant.PATH_SEPARATOR).insert(0, curr.getName());
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public void setFullPath(String fullPath) {
+ this.fullPath = fullPath;
+ }
+
+ @Override
+ public PartialPath getPartialPath() {
+ List<String> detachedPath = new ArrayList<>();
+ IConfigMNode temp = this;
+ detachedPath.add(temp.getName());
+ while (temp.getParent() != null) {
+ temp = temp.getParent();
+ detachedPath.add(0, temp.getName());
+ }
+ return new PartialPath(detachedPath.toArray(new String[0]));
+ }
+
+ @Override
+ public MNodeType getMNodeType() {
+ return MNodeType.SG_INTERNAL;
+ }
+
+ @Override
+ public <R, C> R accept(MNodeVisitor<R, C> visitor, C context) {
+ return visitor.visitBasicMNode(this, context);
+ }
+
+ @Override
+ public int estimateSize() {
+ return 8 + 8 + 8 + 8 + 8 + 8 + 28 + tableNodeInfo.estimateSize();
+ }
+
+ @Override
+ public IConfigMNode getAsMNode() {
+ return this;
+ }
+
+ /** check whether the MNode has a child with the name */
+ @Override
+ public boolean hasChild(String name) {
+ return false;
+ }
+
+ /** get the child with the name */
+ @Override
+ public IConfigMNode getChild(String name) {
+ return null;
+ }
+
+ @Override
+ public IConfigMNode addChild(String name, IConfigMNode child) {
+ return null;
+ }
+
+ @Override
+ public IConfigMNode addChild(IConfigMNode child) {
+ return null;
+ }
+
+ @Override
+ public IConfigMNode deleteChild(String name) {
+ return null;
+ }
+
+ @Override
+ public IMNodeContainer<IConfigMNode> getChildren() {
+ return ConfigMNodeContainer.emptyMNodeContainer();
+ }
+
+ @Override
+ public void setChildren(IMNodeContainer<IConfigMNode> children) {
+ // do nothing
+ }
+
+ @Override
+ public boolean isAboveDatabase() {
+ return false;
+ }
+
+ @Override
+ public boolean isDatabase() {
+ return false;
+ }
+
+ @Override
+ public IDatabaseMNode<IConfigMNode> getAsDatabaseMNode() {
+ throw new UnsupportedOperationException("Wrong MNode Type");
+ }
+
+ @Override
+ public void setSchemaTemplateId(int id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int getSchemaTemplateId() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void preUnsetSchemaTemplate() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void rollbackUnsetSchemaTemplate() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isSchemaTemplatePreUnset() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void unsetSchemaTemplate() {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/impl/TableNodeStatus.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/impl/TableNodeStatus.java
new file mode 100644
index 00000000000..f5787c4e32a
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/impl/TableNodeStatus.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.persistence.schema.mnode.impl;
+
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+public enum TableNodeStatus {
+ PRE_CREATE((byte) 0),
+
+ USING((byte) 1),
+
+ PRE_DELETE((byte) 2);
+
+ private final byte status;
+
+ private TableNodeStatus(byte status) {
+ this.status = status;
+ }
+
+ public void serialize(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(status, outputStream);
+ }
+
+ public static TableNodeStatus deserialize(InputStream inputStream) throws IOException {
+ byte status = ReadWriteIOUtils.readByte(inputStream);
+ switch (status) {
+ case 0:
+ return PRE_CREATE;
+ case 1:
+ return USING;
+ case 2:
+ return PRE_DELETE;
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/info/ConfigTableInfo.java
similarity index 50%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java
copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/info/ConfigTableInfo.java
index a9bca2aef6c..73d1cd2928e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/mnode/info/ConfigTableInfo.java
@@ -17,26 +17,40 @@
* under the License.
*/
-package org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational;
+package org.apache.iotdb.confignode.persistence.schema.mnode.info;
-import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
-import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
-import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
-import org.apache.iotdb.db.relational.sql.tree.CreateTable;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.confignode.persistence.schema.mnode.impl.TableNodeStatus;
+import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.mem.mnode.info.BasicMNodeInfo;
-import com.google.common.util.concurrent.ListenableFuture;
+public class ConfigTableInfo extends BasicMNodeInfo {
-public class CreateTableTask implements IConfigTask {
+ private TsTable table;
- private final CreateTable createTable;
+ private TableNodeStatus status;
- public CreateTableTask(CreateTable createTable) {
- this.createTable = createTable;
+ public ConfigTableInfo(String name) {
+ super(name);
+ }
+
+ public TsTable getTable() {
+ return table;
+ }
+
+ public void setTable(TsTable table) {
+ this.table = table;
+ }
+
+ public TableNodeStatus getStatus() {
+ return status;
+ }
+
+ public void setStatus(TableNodeStatus status) {
+ this.status = status;
}
@Override
- public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
- throws InterruptedException {
- return configTaskExecutor.createTable(createTable);
+ public int estimateSize() {
+ return 1 + 8 + table.getColumnNum() * 30;
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DataNodeRegionTaskExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DataNodeRegionTaskExecutor.java
index b360d4454af..d8bf9cc8d15 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DataNodeRegionTaskExecutor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DataNodeRegionTaskExecutor.java
@@ -63,7 +63,7 @@ public abstract class DataNodeRegionTaskExecutor<Q, R> {
this.dataNodeRequestGenerator = dataNodeRequestGenerator;
}
- void execute() {
+ public void execute() {
// organize region by dataNode
Map<TConsensusGroupId, Set<TDataNodeLocation>> failedHistory = new HashMap<>();
Map<TDataNodeLocation, List<TConsensusGroupId>> dataNodeConsensusGroupIdMap =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java
new file mode 100644
index 00000000000..504b7ff6c6b
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java
@@ -0,0 +1,429 @@
+/*
+ * 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.procedure.impl.schema.table;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathPatternTree;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCType;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
+import org.apache.iotdb.confignode.client.DataNodeRequestType;
+import org.apache.iotdb.confignode.client.async.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler;
+import org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan;
+import org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
+import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.DataNodeRegionTaskExecutor;
+import org.apache.iotdb.confignode.procedure.state.schema.CreateTableState;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.mpp.rpc.thrift.TCheckTimeSeriesExistenceReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCheckTimeSeriesExistenceResp;
+import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import static org.apache.iotdb.commons.conf.IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
+import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
+
+public class CreateTableProcedure
+ extends StateMachineProcedure<ConfigNodeProcedureEnv, CreateTableState> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(CreateTableProcedure.class);
+
+ private String database;
+
+ private TsTable table;
+
+ public CreateTableProcedure() {
+ super();
+ }
+
+ public CreateTableProcedure(String database, TsTable table) {
+ this.database = database;
+ this.table = table;
+ }
+
+ @Override
+ protected Flow executeFromState(ConfigNodeProcedureEnv env, CreateTableState state)
+ throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
+ long startTime = System.currentTimeMillis();
+ try {
+ switch (state) {
+ case PRE_CREATE:
+ LOGGER.info("Pre create table {}.{}", database, table.getTableName());
+ preCreateTable(env);
+ break;
+ case PRE_RELEASE:
+ LOGGER.info("Pre release table {}.{}", database, table.getTableName());
+ preReleaseTable(env);
+ break;
+ case VALIDATE_TIMESERIES_EXISTENCE:
+ LOGGER.info(
+ "Validate timeseries existence for table {}.{}", database, table.getTableName());
+ validateTimeSeriesExistence(env);
+ break;
+ case COMMIT_CREATE:
+ LOGGER.info("Commit create table {}.{}", database, table.getTableName());
+ commitCreateTable(env);
+ break;
+ case COMMIT_RELEASE:
+ LOGGER.info("Commit release table {}.{}", database, table.getTableName());
+ commitReleaseTable(env);
+ return Flow.NO_MORE_STATE;
+ default:
+ setFailure(new ProcedureException("Unrecognized CreateTableState " + state));
+ return Flow.NO_MORE_STATE;
+ }
+ return Flow.HAS_MORE_STATE;
+ } finally {
+ LOGGER.info(
+ "CreateTable-{}.{}-{} costs {}ms",
+ database,
+ table.getTableName(),
+ state,
+ (System.currentTimeMillis() - startTime));
+ }
+ }
+
+ private void preCreateTable(ConfigNodeProcedureEnv env) {
+ PreCreateTablePlan plan = new PreCreateTablePlan(database, table);
+ TSStatus status;
+ try {
+ status = env.getConfigManager().getConsensusManager().write(plan);
+ } catch (ConsensusException e) {
+ LOGGER.warn("Failed in the read API executing the consensus layer due to: ", e);
+ status = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ setNextState(CreateTableState.PRE_RELEASE);
+ } else {
+ setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())));
+ }
+ }
+
+ private void preReleaseTable(ConfigNodeProcedureEnv env) {
+ TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.PRE_CREATE.getOperationType());
+ req.setTableInfo(TsTableInternalRPCUtil.serializeSingleTsTable(database, table));
+
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ AsyncClientHandler<TUpdateTableReq, TSStatus> clientHandler =
+ new AsyncClientHandler<>(DataNodeRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
+ AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);
+ Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ "Failed to sync table {}.{} pre-create info to DataNode {}",
+ database,
+ table.getTableName(),
+ dataNodeLocationMap.get(entry.getKey()));
+ setFailure(new ProcedureException(new MetadataException("Pre create table failed")));
+ return;
+ }
+ }
+ setNextState(CreateTableState.VALIDATE_TIMESERIES_EXISTENCE);
+ }
+
+ private void validateTimeSeriesExistence(ConfigNodeProcedureEnv env) {
+ PathPatternTree patternTree = new PathPatternTree();
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ PartialPath path = null;
+ try {
+ path =
+ new PartialPath(ROOT + PATH_SEPARATOR + database + PATH_SEPARATOR + table.getTableName());
+ patternTree.appendPathPattern(path);
+ patternTree.appendPathPattern(path.concatNode(MULTI_LEVEL_PATH_WILDCARD));
+ patternTree.serialize(dataOutputStream);
+ } catch (IllegalPathException | IOException ignored) {
+ }
+ ByteBuffer patternTreeBytes = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+
+ Map<TConsensusGroupId, TRegionReplicaSet> relatedSchemaRegionGroup =
+ env.getConfigManager().getRelatedSchemaRegionGroup(patternTree);
+
+ List<TCheckTimeSeriesExistenceResp> respList = new ArrayList<>();
+ DataNodeRegionTaskExecutor<TCheckTimeSeriesExistenceReq, TCheckTimeSeriesExistenceResp>
+ regionTask =
+ new DataNodeRegionTaskExecutor<
+ TCheckTimeSeriesExistenceReq, TCheckTimeSeriesExistenceResp>(
+ env,
+ relatedSchemaRegionGroup,
+ false,
+ DataNodeRequestType.CHECK_TIMESERIES_EXISTENCE,
+ ((dataNodeLocation, consensusGroupIdList) ->
+ new TCheckTimeSeriesExistenceReq(patternTreeBytes, consensusGroupIdList))) {
+
+ @Override
+ protected List<TConsensusGroupId> processResponseOfOneDataNode(
+ TDataNodeLocation dataNodeLocation,
+ List<TConsensusGroupId> consensusGroupIdList,
+ TCheckTimeSeriesExistenceResp response) {
+ respList.add(response);
+ List<TConsensusGroupId> failedRegionList = new ArrayList<>();
+ if (response.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return failedRegionList;
+ }
+
+ if (response.getStatus().getCode() == TSStatusCode.MULTIPLE_ERROR.getStatusCode()) {
+ List<TSStatus> subStatus = response.getStatus().getSubStatus();
+ for (int i = 0; i < subStatus.size(); i++) {
+ if (subStatus.get(i).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && subStatus.get(i).getCode()
+ != TSStatusCode.TIMESERIES_ALREADY_EXIST.getStatusCode()) {
+ failedRegionList.add(consensusGroupIdList.get(i));
+ }
+ }
+ } else {
+ failedRegionList.addAll(consensusGroupIdList);
+ }
+ return failedRegionList;
+ }
+
+ @Override
+ protected void onAllReplicasetFailure(
+ TConsensusGroupId consensusGroupId, Set<TDataNodeLocation> dataNodeLocationSet) {
+ setFailure(
+ new ProcedureException(
+ new MetadataException(
+ String.format(
+ "Create table %s.%s failed when [check timeseries existence on DataNode] because all replicaset of schemaRegion %s failed. %s",
+ database,
+ table.getTableName(),
+ consensusGroupId.id,
+ dataNodeLocationSet))));
+ interruptTask();
+ }
+ };
+ regionTask.execute();
+ if (isFailed()) {
+ return;
+ }
+
+ for (TCheckTimeSeriesExistenceResp resp : respList) {
+ if (resp.isExists()) {
+ setFailure(
+ new ProcedureException(
+ new MetadataException(
+ String.format(
+ "Timeseries already exists under root.%s.%s",
+ database, table.getTableName()))));
+ }
+ }
+ setNextState(CreateTableState.COMMIT_CREATE);
+ }
+
+ private void commitCreateTable(ConfigNodeProcedureEnv env) {
+ CommitCreateTablePlan plan = new CommitCreateTablePlan(database, table.getTableName());
+ TSStatus status;
+ try {
+ status = env.getConfigManager().getConsensusManager().write(plan);
+ } catch (ConsensusException e) {
+ LOGGER.warn("Failed in the read API executing the consensus layer due to: ", e);
+ status = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ setNextState(CreateTableState.COMMIT_RELEASE);
+ } else {
+ setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())));
+ }
+ }
+
+ private void commitReleaseTable(ConfigNodeProcedureEnv env) {
+ TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.COMMIT_CREATE.getOperationType());
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try {
+ ReadWriteIOUtils.write(database, outputStream);
+ ReadWriteIOUtils.write(table.getTableName(), outputStream);
+ } catch (IOException ignored) {
+ //
+ }
+ req.setTableInfo(outputStream.toByteArray());
+
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ AsyncClientHandler<TUpdateTableReq, TSStatus> clientHandler =
+ new AsyncClientHandler<>(DataNodeRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
+ AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);
+ Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ "Failed to sync table {}.{} commit-create info to DataNode {}",
+ database,
+ table.getTableName(),
+ dataNodeLocationMap.get(entry.getKey()));
+ setFailure(new ProcedureException(new MetadataException("Commit create table failed")));
+ return;
+ }
+ }
+ }
+
+ @Override
+ protected void rollbackState(ConfigNodeProcedureEnv env, CreateTableState state)
+ throws IOException, InterruptedException, ProcedureException {
+ long startTime = System.currentTimeMillis();
+ try {
+ switch (state) {
+ case PRE_CREATE:
+ LOGGER.info("Start rollback pre create table {}.{}", database, table.getTableName());
+ rollbackCreate(env);
+ break;
+ case PRE_RELEASE:
+ LOGGER.info("Start rollback pre release table {}.{}", database, table.getTableName());
+ rollbackRelease(env);
+ break;
+ }
+ } finally {
+ LOGGER.info(
+ "Rollback CreateTable-{} costs {}ms.", state, (System.currentTimeMillis() - startTime));
+ }
+ }
+
+ private void rollbackCreate(ConfigNodeProcedureEnv env) {
+ RollbackCreateTablePlan plan = new RollbackCreateTablePlan(database, table.getTableName());
+ TSStatus status;
+ try {
+ status = env.getConfigManager().getConsensusManager().write(plan);
+ } catch (ConsensusException e) {
+ LOGGER.warn("Failed in the read API executing the consensus layer due to: ", e);
+ status = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn("Failed to rollback table creation {}.{}", database, table.getTableName());
+ setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())));
+ }
+ }
+
+ private void rollbackRelease(ConfigNodeProcedureEnv env) {
+ TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.ROLLBACK_CREATE.getOperationType());
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try {
+ ReadWriteIOUtils.write(database, outputStream);
+ ReadWriteIOUtils.write(table.getTableName(), outputStream);
+ } catch (IOException ignored) {
+ //
+ }
+ req.setTableInfo(outputStream.toByteArray());
+
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ AsyncClientHandler<TUpdateTableReq, TSStatus> clientHandler =
+ new AsyncClientHandler<>(DataNodeRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
+ AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);
+ Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ "Failed to sync table {}.{} rollback-create info to DataNode {}",
+ database,
+ table.getTableName(),
+ dataNodeLocationMap.get(entry.getKey()));
+ setFailure(new ProcedureException(new MetadataException("Rollback create table failed")));
+ return;
+ }
+ }
+ }
+
+ @Override
+ protected CreateTableState getState(int stateId) {
+ return CreateTableState.values()[stateId];
+ }
+
+ @Override
+ protected int getStateId(CreateTableState createTableState) {
+ return createTableState.ordinal();
+ }
+
+ @Override
+ protected CreateTableState getInitialState() {
+ return CreateTableState.PRE_CREATE;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public TsTable getTable() {
+ return table;
+ }
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ stream.writeShort(ProcedureType.CREATE_TABLE_PROCEDURE.getTypeCode());
+ super.serialize(stream);
+ ReadWriteIOUtils.write(database, stream);
+ table.serialize(stream);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer byteBuffer) {
+ super.deserialize(byteBuffer);
+ database = ReadWriteIOUtils.readString(byteBuffer);
+ table = TsTable.deserialize(byteBuffer);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof CreateTableProcedure)) return false;
+ CreateTableProcedure that = (CreateTableProcedure) o;
+ return Objects.equals(database, that.database) && Objects.equals(table, that.table);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(database, table);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/schema/CreateTableState.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/schema/CreateTableState.java
new file mode 100644
index 00000000000..ba22c4a892d
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/schema/CreateTableState.java
@@ -0,0 +1,28 @@
+/*
+ * 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.procedure.state.schema;
+
+public enum CreateTableState {
+ PRE_CREATE,
+ PRE_RELEASE,
+ VALIDATE_TIMESERIES_EXISTENCE,
+ COMMIT_CREATE,
+ COMMIT_RELEASE,
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
index 7f0d8245660..394a76a802d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
@@ -45,6 +45,7 @@ import org.apache.iotdb.confignode.procedure.impl.schema.DeleteLogicalViewProced
import org.apache.iotdb.confignode.procedure.impl.schema.DeleteTimeSeriesProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.SetTemplateProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.UnsetTemplateProcedure;
+import org.apache.iotdb.confignode.procedure.impl.schema.table.CreateTableProcedure;
import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.AlterConsumerGroupProcedure;
import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.CreateConsumerProcedure;
import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.DropConsumerProcedure;
@@ -177,6 +178,9 @@ public class ProcedureFactory implements IProcedureFactory {
case UNSET_TEMPLATE_PROCEDURE:
procedure = new UnsetTemplateProcedure(false);
break;
+ case CREATE_TABLE_PROCEDURE:
+ procedure = new CreateTableProcedure();
+ break;
case CREATE_PIPE_PLUGIN_PROCEDURE:
procedure = new CreatePipePluginProcedure();
break;
@@ -302,6 +306,8 @@ public class ProcedureFactory implements IProcedureFactory {
return ProcedureType.DEACTIVATE_TEMPLATE_PROCEDURE;
} else if (procedure instanceof UnsetTemplateProcedure) {
return ProcedureType.UNSET_TEMPLATE_PROCEDURE;
+ } else if (procedure instanceof CreateTableProcedure) {
+ return ProcedureType.CREATE_TABLE_PROCEDURE;
} else if (procedure instanceof CreatePipePluginProcedure) {
return ProcedureType.CREATE_PIPE_PLUGIN_PROCEDURE;
} else if (procedure instanceof DropPipePluginProcedure) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
index 1711953e741..39a92287cdf 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
@@ -62,6 +62,8 @@ public enum ProcedureType {
UNSET_TEMPLATE_PROCEDURE((short) 701),
SET_TEMPLATE_PROCEDURE((short) 702),
+ CREATE_TABLE_PROCEDURE((short) 750),
+
// ProcedureId 800-899 is used by IoTDB-Ml
/** Pipe Plugin */
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index caa0afcb75b..e88ab55d144 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -1109,4 +1109,9 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
public TThrottleQuotaResp getThrottleQuota() {
return configManager.getThrottleQuota();
}
+
+ @Override
+ public TSStatus createTable(ByteBuffer tableInfo) throws TException {
+ return configManager.createTable(tableInfo);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/metadata/table/TableAlreadyExistsException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/metadata/table/TableAlreadyExistsException.java
new file mode 100644
index 00000000000..d745b6eb0a3
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/metadata/table/TableAlreadyExistsException.java
@@ -0,0 +1,32 @@
+/*
+ * 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.db.exception.metadata.table;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class TableAlreadyExistsException extends MetadataException {
+
+ public TableAlreadyExistsException(String database, String tableName) {
+ super(
+ String.format("Table %s.%s already exists.", database, tableName),
+ TSStatusCode.TABLE_ALREADY_EXISTS.getStatusCode());
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/metadata/table/TableNotExistsException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/metadata/table/TableNotExistsException.java
new file mode 100644
index 00000000000..35ec8873ba4
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/metadata/table/TableNotExistsException.java
@@ -0,0 +1,32 @@
+/*
+ * 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.db.exception.metadata.table;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class TableNotExistsException extends MetadataException {
+
+ public TableNotExistsException(String database, String tableName) {
+ super(
+ String.format("Table %s.%s not exists.", database, tableName),
+ TSStatusCode.TABLE_NOT_EXISTS.getStatusCode());
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index f62fda39de3..f31728aa1b8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -150,6 +150,7 @@ import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
@@ -1066,6 +1067,12 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
() -> client.getThrottleQuota(), resp -> !updateConfigNodeLeader(resp.status));
}
+ @Override
+ public TSStatus createTable(ByteBuffer tableInfo) throws TException {
+ return executeRemoteCallWithRetry(
+ () -> client.createTable(tableInfo), status -> !updateConfigNodeLeader(status));
+ }
+
public static class Factory extends ThriftClientFactory<ConfigRegionId, ConfigNodeClient> {
public Factory(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index fb3e2412e4a..d1df0a2638a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -45,6 +45,9 @@ import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.pipe.plugin.meta.PipePluginMeta;
import org.apache.iotdb.commons.pipe.task.meta.PipeMeta;
import org.apache.iotdb.commons.schema.SchemaConstant;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCType;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -122,6 +125,7 @@ import org.apache.iotdb.db.schemaengine.SchemaEngine;
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.info.ITimeSeriesSchemaInfo;
import org.apache.iotdb.db.schemaengine.schemaregion.read.resp.reader.ISchemaReader;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.schemaengine.template.TemplateInternalRPCUpdateType;
import org.apache.iotdb.db.service.DataNode;
@@ -218,6 +222,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp;
import org.apache.iotdb.mpp.rpc.thrift.TTsFilePieceReq;
+import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTriggerLocationReq;
import org.apache.iotdb.rpc.RpcUtils;
@@ -228,6 +233,7 @@ import org.apache.iotdb.trigger.api.enums.TriggerEvent;
import org.apache.iotdb.tsfile.exception.NotImplementedException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.utils.RamUsageEstimator;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.iotdb.tsfile.write.record.Tablet;
@@ -1378,6 +1384,39 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
return resp;
}
+ @Override
+ public TSStatus updateTable(TUpdateTableReq req) throws TException {
+ switch (TsTableInternalRPCType.getType(req.type)) {
+ case PRE_CREATE:
+ DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.TIMESERIES_VS_TABLE);
+ try {
+ Pair<String, TsTable> pair =
+ TsTableInternalRPCUtil.deserializeSingleTsTable(req.getTableInfo());
+ DataNodeTableCache.getInstance().preCreateTable(pair.left, pair.right);
+ break;
+ } finally {
+ DataNodeSchemaLockManager.getInstance()
+ .releaseWriteLock(SchemaLockType.TIMESERIES_VS_TABLE);
+ }
+ case ROLLBACK_CREATE:
+ DataNodeTableCache.getInstance()
+ .rollbackCreateTable(
+ ReadWriteIOUtils.readString(req.tableInfo),
+ ReadWriteIOUtils.readString(req.tableInfo));
+ break;
+ case COMMIT_CREATE:
+ DataNodeTableCache.getInstance()
+ .commitCreateTable(
+ ReadWriteIOUtils.readString(req.tableInfo),
+ ReadWriteIOUtils.readString(req.tableInfo));
+ break;
+ default:
+ LOGGER.warn("Unsupported type {} when updating table", req.type);
+ return RpcUtils.getStatus(TSStatusCode.ILLEGAL_PARAMETER);
+ }
+ return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ }
+
private PathPatternTree filterPathPatternTree(PathPatternTree patternTree, String storageGroup) {
PathPatternTree filteredPatternTree = new PathPatternTree();
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
index 059a4a218b5..af3d2a949cb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/ColumnHeaderConstant.java
@@ -200,6 +200,12 @@ public class ColumnHeaderConstant {
// column names for show current timestamp
public static final String CURRENT_TIMESTAMP = "CurrentTimestamp";
+ // column names for table query
+ public static final String COLUMN_NAME = "ColumnName";
+ public static final String COLUMN_DATA_TYPE = "DataType";
+ public static final String COLUMN_CATEGORY = "Category";
+ public static final String TABLE_NAME = "TableName";
+
public static final List<ColumnHeader> lastQueryColumnHeaders =
ImmutableList.of(
new ColumnHeader(TIMESERIES, TSDataType.TEXT),
@@ -496,4 +502,15 @@ public class ColumnHeaderConstant {
new ColumnHeader(SCHEMA_REPLICATION_FACTOR, TSDataType.INT32),
new ColumnHeader(DATA_REPLICATION_FACTOR, TSDataType.INT32),
new ColumnHeader(TIME_PARTITION_INTERVAL, TSDataType.INT64));
+
+ public static final List<ColumnHeader> describeTableColumnHeaders =
+ ImmutableList.of(
+ new ColumnHeader(COLUMN_NAME, TSDataType.TEXT),
+ new ColumnHeader(COLUMN_DATA_TYPE, TSDataType.TEXT),
+ new ColumnHeader(COLUMN_CATEGORY, TSDataType.TEXT));
+
+ public static final List<ColumnHeader> showTablesColumnHeaders =
+ ImmutableList.of(
+ new ColumnHeader(TABLE_NAME, TSDataType.TEXT),
+ new ColumnHeader(COLUMN_TTL, TSDataType.INT64));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
index ae684cec09e..16bb695ecb7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/header/DatasetHeaderFactory.java
@@ -208,4 +208,12 @@ public class DatasetHeaderFactory {
public static DatasetHeader getShowDBHeader() {
return new DatasetHeader(ColumnHeaderConstant.showDBColumnHeaders, true);
}
+
+ public static DatasetHeader getDescribeTableHeader() {
+ return new DatasetHeader(ColumnHeaderConstant.describeTableColumnHeaders, true);
+ }
+
+ public static DatasetHeader getShowTablesHeader() {
+ return new DatasetHeader(ColumnHeaderConstant.showTablesColumnHeaders, true);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index 966c6a21df4..6a977a196fb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -275,7 +275,7 @@ public class Coordinator {
queryContext,
null,
executor,
- statement.accept(new TableConfigTaskVisitor(clientSession), queryContext));
+ statement.accept(new TableConfigTaskVisitor(clientSession, metadata), queryContext));
}
RelationalModelPlanner treeModelPlanner =
new RelationalModelPlanner(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
index 2638c29c257..2a681119b67 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
@@ -40,6 +40,7 @@ import org.apache.iotdb.commons.utils.TimePartitionUtils;
import org.apache.iotdb.confignode.rpc.thrift.TGetDataNodeLocationsResp;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.metadata.table.TableAlreadyExistsException;
import org.apache.iotdb.db.exception.metadata.template.TemplateIncompatibleException;
import org.apache.iotdb.db.exception.metadata.view.UnsupportedViewException;
import org.apache.iotdb.db.exception.sql.SemanticException;
@@ -139,6 +140,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainAnalyzeStatemen
import org.apache.iotdb.db.queryengine.plan.statement.sys.ExplainStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.ShowQueriesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.ShowVersionStatement;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.schemaengine.template.Template;
import org.apache.iotdb.db.utils.constant.SqlConstant;
import org.apache.iotdb.rpc.RpcUtils;
@@ -2299,6 +2301,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Analysis analysis = new Analysis();
analysis.setStatement(createTimeSeriesStatement);
+ checkIsTableCompatible(createTimeSeriesStatement.getPath(), context);
checkIsTemplateCompatible(
createTimeSeriesStatement.getPath(), createTimeSeriesStatement.getAlias(), context);
@@ -2346,6 +2349,16 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
}
}
+ private void checkIsTableCompatible(PartialPath timeseriesPath, MPPQueryContext context) {
+ DataNodeSchemaLockManager.getInstance().takeReadLock(SchemaLockType.TIMESERIES_VS_TABLE);
+ context.addAcquiredLockNum(SchemaLockType.TIMESERIES_VS_TABLE);
+ Pair<String, String> tableInfo =
+ DataNodeTableCache.getInstance().checkTableCreateAndPreCreateOnGivenPath(timeseriesPath);
+ if (tableInfo != null) {
+ throw new SemanticException(new TableAlreadyExistsException(tableInfo.left, tableInfo.right));
+ }
+ }
+
private void analyzeSchemaProps(Map<String, String> props) {
if (props == null || props.isEmpty()) {
return;
@@ -2395,6 +2408,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Analysis analysis = new Analysis();
analysis.setStatement(createAlignedTimeSeriesStatement);
+ checkIsTableCompatible(createAlignedTimeSeriesStatement.getDevicePath(), context);
checkIsTemplateCompatible(
createAlignedTimeSeriesStatement.getDevicePath(),
createAlignedTimeSeriesStatement.getMeasurements(),
@@ -2422,6 +2436,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
Analysis analysis = new Analysis();
analysis.setStatement(internalCreateTimeSeriesStatement);
+ checkIsTableCompatible(internalCreateTimeSeriesStatement.getDevicePath(), context);
checkIsTemplateCompatible(
internalCreateTimeSeriesStatement.getDevicePath(),
internalCreateTimeSeriesStatement.getMeasurements(),
@@ -2454,6 +2469,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
PathPatternTree pathPatternTree = new PathPatternTree();
for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry :
internalCreateMultiTimeSeriesStatement.getDeviceMap().entrySet()) {
+ checkIsTableCompatible(entry.getKey(), context);
checkIsTemplateCompatible(
entry.getKey(), entry.getValue().right.getMeasurements(), null, context);
pathPatternTree.appendFullPath(entry.getKey().concatNode(ONE_LEVEL_PATH_WILDCARD));
@@ -2479,6 +2495,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
List<PartialPath> timeseriesPathList = createMultiTimeSeriesStatement.getPaths();
List<String> aliasList = createMultiTimeSeriesStatement.getAliasList();
for (int i = 0; i < timeseriesPathList.size(); i++) {
+ checkIsTableCompatible(timeseriesPathList.get(i), context);
checkIsTemplateCompatible(
timeseriesPathList.get(i), aliasList == null ? null : aliasList.get(i), context);
}
@@ -3589,6 +3606,7 @@ public class AnalyzeVisitor extends StatementVisitor<Analysis, MPPQueryContext>
// Make sure all paths are not under any templates
try {
for (PartialPath path : createLogicalViewStatement.getTargetPathList()) {
+ checkIsTableCompatible(path, context);
checkIsTemplateCompatible(path, null, context);
}
} catch (Exception e) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java
index b72a67c1fac..6fa4d11d0a1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/lock/SchemaLockType.java
@@ -30,7 +30,7 @@ public enum SchemaLockType {
*
* <ol>
* <li>Take read lock before creating timeseries.
- * <li>Release read lock after finishing creating timeseries
+ * <li>Release read lock after finishing creating timeseries.
* <li>Take write lock before pre-set template.
* <li>Release write lock after finishing pre-set template.
* </ol>
@@ -47,5 +47,18 @@ public enum SchemaLockType {
* <li>Release write lock after finishing invalidating schema cache.
* </ol>
*/
- VALIDATE_VS_DELETION
+ VALIDATE_VS_DELETION,
+
+ /**
+ * This lock is used for guarantee no timeseries (tree model) under path representing table (table
+ * model).
+ *
+ * <ol>
+ * <li>Take read lock before creating timeseries (tree model).
+ * <li>Release read lock after finishing creating timeseries (tree model).
+ * <li>Take write lock before pre-create table (table model).
+ * <li>Release write lock after finishing pre-create table (table model).
+ * </ol>
+ */
+ TIMESERIES_VS_TABLE,
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index 65c045b5dd7..3ac52823be0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -19,18 +19,29 @@
package org.apache.iotdb.db.queryengine.plan.execution.config;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.IdColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.MeasurementColumnSchema;
+import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.CreateDBTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.CreateTableTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DescribeTableTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DropDBTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowDBTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowTablesTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.UseDBTask;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.type.TypeNotFoundException;
import org.apache.iotdb.db.relational.sql.tree.AstVisitor;
+import org.apache.iotdb.db.relational.sql.tree.ColumnDefinition;
import org.apache.iotdb.db.relational.sql.tree.CreateDB;
import org.apache.iotdb.db.relational.sql.tree.CreateTable;
import org.apache.iotdb.db.relational.sql.tree.CurrentDatabase;
+import org.apache.iotdb.db.relational.sql.tree.DataType;
import org.apache.iotdb.db.relational.sql.tree.DescribeTable;
import org.apache.iotdb.db.relational.sql.tree.DropDB;
import org.apache.iotdb.db.relational.sql.tree.DropTable;
@@ -38,13 +49,23 @@ import org.apache.iotdb.db.relational.sql.tree.Node;
import org.apache.iotdb.db.relational.sql.tree.ShowDB;
import org.apache.iotdb.db.relational.sql.tree.ShowTables;
import org.apache.iotdb.db.relational.sql.tree.Use;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.type.Type;
+import org.apache.iotdb.tsfile.read.common.type.TypeEnum;
+
+import static org.apache.iotdb.db.queryengine.plan.relational.type.TypeSignatureTranslator.toTypeSignature;
+import static org.apache.iotdb.db.utils.EncodingInferenceUtils.getDefaultEncoding;
public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryContext> {
private final IClientSession clientSession;
- public TableConfigTaskVisitor(IClientSession clientSession) {
+ private final Metadata metadata;
+
+ public TableConfigTaskVisitor(IClientSession clientSession, Metadata metadata) {
this.clientSession = clientSession;
+ this.metadata = metadata;
}
@Override
@@ -79,10 +100,70 @@ public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryCont
@Override
protected IConfigTask visitCreateTable(CreateTable node, MPPQueryContext context) {
- String currentDatabase = clientSession.getDatabaseName();
- // TODO using currentDatabase to normalize QualifiedName in CreateTable
+ context.setQueryType(QueryType.WRITE);
+ String database = clientSession.getDatabaseName();
+ if (node.getName().getPrefix().isPresent()) {
+ database = node.getName().getPrefix().get().toString();
+ }
+ if (database == null) {
+ throw new SemanticException("unknown database");
+ }
+ TsTable table = new TsTable(node.getName().getSuffix());
+ for (ColumnDefinition columnDefinition : node.getElements()) {
+ ColumnDefinition.ColumnCategory category = columnDefinition.getColumnCategory();
+ String columnName = columnDefinition.getName().getValue();
+ if (table.getColumnSchema(columnName) != null) {
+ throw new SemanticException(
+ String.format("Columns in table shall not share the same name %s.", columnName));
+ }
+ TSDataType dataType = getDataType(columnDefinition.getType());
+ switch (category) {
+ case ID:
+ table.addColumnSchema(new IdColumnSchema(columnName, dataType));
+ break;
+ case ATTRIBUTE:
+ table.addColumnSchema(new AttributeColumnSchema(columnName, dataType));
+ break;
+ case TIME:
+ break;
+ case MEASUREMENT:
+ table.addColumnSchema(
+ new MeasurementColumnSchema(
+ columnName,
+ dataType,
+ getDefaultEncoding(dataType),
+ TSFileDescriptor.getInstance().getConfig().getCompressor()));
+ break;
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+ return new CreateTableTask(table, database, node.isIfNotExists());
+ }
- return new CreateTableTask(node);
+ private TSDataType getDataType(DataType dataType) {
+ try {
+ Type type = metadata.getType(toTypeSignature(dataType));
+ TypeEnum typeEnum = type.getTypeEnum();
+ switch (typeEnum) {
+ case TEXT:
+ return TSDataType.TEXT;
+ case FLOAT:
+ return TSDataType.FLOAT;
+ case DOUBLE:
+ return TSDataType.DOUBLE;
+ case INT32:
+ return TSDataType.INT32;
+ case INT64:
+ return TSDataType.INT64;
+ case BOOLEAN:
+ return TSDataType.BOOLEAN;
+ default:
+ throw new IllegalArgumentException();
+ }
+ } catch (TypeNotFoundException e) {
+ throw new SemanticException(String.format("Unknown type: %s", dataType));
+ }
}
@Override
@@ -92,12 +173,28 @@ public class TableConfigTaskVisitor extends AstVisitor<IConfigTask, MPPQueryCont
@Override
protected IConfigTask visitShowTables(ShowTables node, MPPQueryContext context) {
- return super.visitShowTables(node, context);
+ context.setQueryType(QueryType.READ);
+ String database = clientSession.getDatabaseName();
+ if (node.getDbName().isPresent()) {
+ database = node.getDbName().get().toString();
+ }
+ if (database == null) {
+ throw new SemanticException("unknown database");
+ }
+ return new ShowTablesTask(database);
}
@Override
protected IConfigTask visitDescribeTable(DescribeTable node, MPPQueryContext context) {
- return super.visitDescribeTable(node, context);
+ context.setQueryType(QueryType.READ);
+ String database = clientSession.getDatabaseName();
+ if (node.getTable().getPrefix().isPresent()) {
+ database = node.getTable().getPrefix().get().toString();
+ }
+ if (database == null) {
+ throw new SemanticException("unknown database");
+ }
+ return new DescribeTableTask(database, node.getTable().getSuffix());
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 3286a256695..e13d5baae4f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -42,6 +42,8 @@ import org.apache.iotdb.commons.pipe.connector.payload.airgap.AirGapPseudoTPipeT
import org.apache.iotdb.commons.pipe.plugin.service.PipePluginClassLoader;
import org.apache.iotdb.commons.pipe.plugin.service.PipePluginExecutableManager;
import org.apache.iotdb.commons.pipe.task.meta.PipeStaticMeta;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
import org.apache.iotdb.commons.schema.view.LogicalViewSchema;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
@@ -108,6 +110,7 @@ import org.apache.iotdb.db.exception.BatchProcessException;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
+import org.apache.iotdb.db.exception.metadata.table.TableNotExistsException;
import org.apache.iotdb.db.exception.sql.SemanticException;
import org.apache.iotdb.db.pipe.agent.PipeAgent;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
@@ -142,7 +145,9 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowRegion
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowTTLTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowTriggersTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.ShowVariablesTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.DescribeTableTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowDBTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowTablesTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.template.ShowNodesInSchemaTemplateTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.template.ShowPathSetTemplateTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.template.ShowSchemaTemplateTask;
@@ -202,12 +207,12 @@ import org.apache.iotdb.db.queryengine.plan.statement.sys.quota.SetThrottleQuota
import org.apache.iotdb.db.queryengine.plan.statement.sys.quota.ShowSpaceQuotaStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.quota.ShowThrottleQuotaStatement;
import org.apache.iotdb.db.relational.sql.tree.CreateDB;
-import org.apache.iotdb.db.relational.sql.tree.CreateTable;
import org.apache.iotdb.db.relational.sql.tree.DropDB;
import org.apache.iotdb.db.relational.sql.tree.ShowDB;
import org.apache.iotdb.db.relational.sql.tree.Use;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
import org.apache.iotdb.db.schemaengine.rescon.DataNodeSchemaQuotaManager;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.schemaengine.template.Template;
import org.apache.iotdb.db.schemaengine.template.TemplateAlterOperationType;
@@ -2789,8 +2794,66 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
}
@Override
- public SettableFuture<ConfigTaskResult> createTable(CreateTable createTable) {
- return null;
+ public SettableFuture<ConfigTaskResult> createTable(
+ TsTable table, String database, boolean ifNotExists) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try (ConfigNodeClient configNodeClient =
+ CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+
+ TSStatus tsStatus;
+ do {
+ try {
+ tsStatus =
+ configNodeClient.createTable(
+ ByteBuffer.wrap(TsTableInternalRPCUtil.serializeSingleTsTable(database, table)));
+ } catch (TTransportException e) {
+ if (e.getType() == TTransportException.TIMED_OUT
+ || e.getCause() instanceof SocketTimeoutException) {
+ // Time out mainly caused by slow execution, just wait
+ tsStatus = RpcUtils.getStatus(TSStatusCode.OVERLAP_WITH_EXISTING_TASK);
+ } else {
+ throw e;
+ }
+ }
+ // Keep waiting until task ends
+ } while (TSStatusCode.OVERLAP_WITH_EXISTING_TASK.getStatusCode() == tsStatus.getCode());
+
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() == tsStatus.getCode()
+ || (TSStatusCode.TABLE_ALREADY_EXISTS.getStatusCode() == tsStatus.getCode()
+ && ifNotExists)) {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ } else {
+ LOGGER.warn(
+ "Failed to create table {}.{} in config node, status is {}.",
+ database,
+ table.getTableName(),
+ tsStatus);
+ future.setException(new IoTDBException(tsStatus.getMessage(), tsStatus.getCode()));
+ }
+ } catch (ClientManagerException | TException e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> describeTable(String database, String tableName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName);
+ if (table == null) {
+ future.setException(new TableNotExistsException(database, tableName));
+ } else {
+ DescribeTableTask.buildTsBlock(table, future);
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showTables(String database) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ List<TsTable> tableList = DataNodeTableCache.getInstance().getTables(database);
+ ShowTablesTask.buildTsBlock(tableList, future);
+ return future;
}
public void handlePipeConfigClientExit(String clientId) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
index 91c5ea88884..b5f8ce9c54f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.plan.execution.config.executor;
import org.apache.iotdb.common.rpc.thrift.TFlushReq;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp;
import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp;
import org.apache.iotdb.db.protocol.session.IClientSession;
@@ -75,7 +76,6 @@ import org.apache.iotdb.db.queryengine.plan.statement.sys.quota.SetThrottleQuota
import org.apache.iotdb.db.queryengine.plan.statement.sys.quota.ShowSpaceQuotaStatement;
import org.apache.iotdb.db.queryengine.plan.statement.sys.quota.ShowThrottleQuotaStatement;
import org.apache.iotdb.db.relational.sql.tree.CreateDB;
-import org.apache.iotdb.db.relational.sql.tree.CreateTable;
import org.apache.iotdb.db.relational.sql.tree.DropDB;
import org.apache.iotdb.db.relational.sql.tree.ShowDB;
import org.apache.iotdb.db.relational.sql.tree.Use;
@@ -256,5 +256,9 @@ public interface IConfigTaskExecutor {
SettableFuture<ConfigTaskResult> createDatabase(CreateDB createDB);
- SettableFuture<ConfigTaskResult> createTable(CreateTable createTable);
+ SettableFuture<ConfigTaskResult> createTable(TsTable table, String database, boolean ifNotExists);
+
+ SettableFuture<ConfigTaskResult> describeTable(String database, String tableName);
+
+ SettableFuture<ConfigTaskResult> showTables(String database);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java
index a9bca2aef6c..c0e62f4a183 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/CreateTableTask.java
@@ -19,24 +19,30 @@
package org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational;
+import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
-import org.apache.iotdb.db.relational.sql.tree.CreateTable;
import com.google.common.util.concurrent.ListenableFuture;
public class CreateTableTask implements IConfigTask {
- private final CreateTable createTable;
+ private final TsTable table;
- public CreateTableTask(CreateTable createTable) {
- this.createTable = createTable;
+ private final String database;
+
+ private final boolean ifNotExists;
+
+ public CreateTableTask(TsTable table, String database, boolean ifNotExists) {
+ this.table = table;
+ this.database = database;
+ this.ifNotExists = ifNotExists;
}
@Override
public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
throws InterruptedException {
- return configTaskExecutor.createTable(createTable);
+ return configTaskExecutor.createTable(table, database, ifNotExists);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/DescribeTableTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/DescribeTableTask.java
new file mode 100644
index 00000000000..8b33ae4e428
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/DescribeTableTask.java
@@ -0,0 +1,85 @@
+/*
+ * 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.db.queryengine.plan.execution.config.metadata.relational;
+
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
+import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
+import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class DescribeTableTask implements IConfigTask {
+
+ private final String database;
+
+ private final String tableName;
+
+ public DescribeTableTask(String database, String tableName) {
+ this.database = database;
+ this.tableName = tableName;
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.describeTable(database, tableName);
+ }
+
+ public static void buildTsBlock(TsTable table, SettableFuture<ConfigTaskResult> future) {
+ List<TSDataType> outputDataTypes =
+ ColumnHeaderConstant.describeTableColumnHeaders.stream()
+ .map(ColumnHeader::getColumnType)
+ .collect(Collectors.toList());
+
+ TsBlockBuilder builder = new TsBlockBuilder(outputDataTypes);
+ for (TsTableColumnSchema columnSchema : table.getColumnList()) {
+ builder.getTimeColumnBuilder().writeLong(0L);
+ builder
+ .getColumnBuilder(0)
+ .writeBinary(new Binary(columnSchema.getColumnName(), TSFileConfig.STRING_CHARSET));
+ builder
+ .getColumnBuilder(1)
+ .writeBinary(new Binary(columnSchema.getDataType().name(), TSFileConfig.STRING_CHARSET));
+ builder
+ .getColumnBuilder(2)
+ .writeBinary(
+ new Binary(columnSchema.getColumnCategory().name(), TSFileConfig.STRING_CHARSET));
+ builder.declarePosition();
+ }
+
+ DatasetHeader datasetHeader = DatasetHeaderFactory.getDescribeTableHeader();
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS, builder.build(), datasetHeader));
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowTablesTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowTablesTask.java
new file mode 100644
index 00000000000..586f0d0e744
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowTablesTask.java
@@ -0,0 +1,77 @@
+/*
+ * 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.db.queryengine.plan.execution.config.metadata.relational;
+
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
+import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
+import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.iotdb.tsfile.utils.Binary;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class ShowTablesTask implements IConfigTask {
+
+ private final String database;
+
+ public ShowTablesTask(String database) {
+ this.database = database;
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.showTables(database);
+ }
+
+ public static void buildTsBlock(
+ List<TsTable> tableList, SettableFuture<ConfigTaskResult> future) {
+ List<TSDataType> outputDataTypes =
+ ColumnHeaderConstant.showTablesColumnHeaders.stream()
+ .map(ColumnHeader::getColumnType)
+ .collect(Collectors.toList());
+
+ TsBlockBuilder builder = new TsBlockBuilder(outputDataTypes);
+ for (TsTable table : tableList) {
+ builder.getTimeColumnBuilder().writeLong(0L);
+
+ builder
+ .getColumnBuilder(0)
+ .writeBinary(new Binary(table.getTableName(), TSFileConfig.STRING_CHARSET));
+ builder.getColumnBuilder(1).appendNull();
+ builder.declarePosition();
+ }
+
+ DatasetHeader datasetHeader = DatasetHeaderFactory.getShowTablesHeader();
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS, builder.build(), datasetHeader));
+ }
+}
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
new file mode 100644
index 00000000000..83bfd434f82
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -0,0 +1,207 @@
+/*
+ * 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.db.schemaengine.table;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_ROOT;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.PATH_SEPARATOR;
+
+public class DataNodeTableCache implements ITableCache {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeTableCache.class);
+
+ private final Map<String, Map<String, TsTable>> databaseTableMap = new ConcurrentHashMap<>();
+
+ private final Map<String, Map<String, TsTable>> preCreateTableMap = new ConcurrentHashMap<>();
+
+ private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+
+ private DataNodeTableCache() {
+ // do nothing
+ }
+
+ private static final class DataNodeTableCacheHolder {
+ private static final DataNodeTableCache INSTANCE = new DataNodeTableCache();
+
+ private DataNodeTableCacheHolder() {}
+ }
+
+ public static DataNodeTableCache getInstance() {
+ return DataNodeTableCacheHolder.INSTANCE;
+ }
+
+ @Override
+ public void init(byte[] tableInitializationBytes) {
+ readWriteLock.writeLock().lock();
+ try {
+ if (tableInitializationBytes == null) {
+ return;
+ }
+ Pair<Map<String, List<TsTable>>, Map<String, List<TsTable>>> tableInfo =
+ TsTableInternalRPCUtil.deserializeTableInitializationInfo(tableInitializationBytes);
+ Map<String, List<TsTable>> usingMap = tableInfo.left;
+ Map<String, List<TsTable>> preCreateMap = tableInfo.right;
+ saveUpdatedTableInfo(usingMap, databaseTableMap);
+ saveUpdatedTableInfo(preCreateMap, preCreateTableMap);
+ LOGGER.info("Init DataNodeTableCache successfully");
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ private void saveUpdatedTableInfo(
+ Map<String, List<TsTable>> tableMap, Map<String, Map<String, TsTable>> localTableMap) {
+ for (Map.Entry<String, List<TsTable>> entry : tableMap.entrySet()) {
+ Map<String, TsTable> map = new ConcurrentHashMap<>();
+ for (TsTable table : entry.getValue()) {
+ map.put(table.getTableName(), table);
+ }
+ localTableMap.put(entry.getKey(), map);
+ }
+ }
+
+ @Override
+ public void preCreateTable(String database, TsTable table) {
+ readWriteLock.writeLock().lock();
+ try {
+ preCreateTableMap
+ .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
+ .put(table.getTableName(), table);
+ LOGGER.info("Pre-create table {}.{} successfully", database, table);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ @Override
+ public void rollbackCreateTable(String database, String tableName) {
+ readWriteLock.writeLock().lock();
+ try {
+ removeTableFromPreCreateMap(database, tableName);
+ LOGGER.info("Rollback-create table {}.{} successfully", database, tableName);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ private void removeTableFromPreCreateMap(String database, String tableName) {
+ preCreateTableMap.compute(
+ database,
+ (k, v) -> {
+ if (v == null) {
+ throw new IllegalStateException();
+ }
+ v.remove(tableName);
+ if (v.isEmpty()) {
+ return null;
+ } else {
+ return v;
+ }
+ });
+ }
+
+ @Override
+ public void commitCreateTable(String database, String tableName) {
+ readWriteLock.writeLock().lock();
+ try {
+ TsTable table = preCreateTableMap.get(database).get(tableName);
+ databaseTableMap
+ .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
+ .put(tableName, table);
+ removeTableFromPreCreateMap(database, tableName);
+ LOGGER.info("Commit-create table {}.{} successfully", database, tableName);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ public TsTable getTable(String database, String tableName) {
+ readWriteLock.readLock().lock();
+ try {
+ if (databaseTableMap.containsKey(database)) {
+ return databaseTableMap.get(database).get(tableName);
+ }
+ return null;
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ public List<TsTable> getTables(String database) {
+ readWriteLock.readLock().lock();
+ try {
+ if (databaseTableMap.containsKey(database)) {
+ return new ArrayList<>(databaseTableMap.get(database).values());
+ }
+ return Collections.emptyList();
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ /** Check whether the given path overlap with some table existence. */
+ public Pair<String, String> checkTableCreateAndPreCreateOnGivenPath(PartialPath path) {
+ readWriteLock.writeLock().lock();
+ try {
+ String pathString = path.getFullPath();
+ Pair<String, String> result = checkTableExistenceOnGivenPath(pathString, databaseTableMap);
+ if (result == null) {
+ result = checkTableExistenceOnGivenPath(pathString, preCreateTableMap);
+ }
+ return result;
+ } finally {
+ readWriteLock.writeLock().unlock();
+ }
+ }
+
+ private Pair<String, String> checkTableExistenceOnGivenPath(
+ String path, Map<String, Map<String, TsTable>> tableMap) {
+ int dbStartIndex = PATH_ROOT.length() + 1;
+ for (Map.Entry<String, Map<String, TsTable>> dbEntry : tableMap.entrySet()) {
+ String database = dbEntry.getKey();
+ if (!(path.startsWith(database, dbStartIndex)
+ && path.charAt(dbStartIndex + database.length()) == PATH_SEPARATOR)) {
+ continue;
+ }
+ int tableStartIndex = dbStartIndex + database.length() + 1;
+ for (String tableName : dbEntry.getValue().keySet()) {
+ if (path.startsWith(tableName, tableStartIndex)
+ && path.charAt(tableStartIndex + tableName.length()) == PATH_SEPARATOR) {
+ return new Pair<>(database, tableName);
+ }
+ }
+ }
+ return null;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
new file mode 100644
index 00000000000..16d26300473
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java
@@ -0,0 +1,33 @@
+/*
+ * 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.db.schemaengine.table;
+
+import org.apache.iotdb.commons.schema.table.TsTable;
+
+public interface ITableCache {
+
+ void init(byte[] tableInitializationBytes);
+
+ void preCreateTable(String database, TsTable table);
+
+ void rollbackCreateTable(String database, String tableName);
+
+ void commitCreateTable(String database, String tableName);
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index e9752f828c7..efc77fdb7d0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -81,6 +81,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.distribution.DistributionPla
import org.apache.iotdb.db.queryengine.plan.planner.distribution.SourceRewriter;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
import org.apache.iotdb.db.service.metrics.DataNodeMetricsHelper;
import org.apache.iotdb.db.service.metrics.IoTDBInternalLocalReporter;
@@ -386,6 +387,9 @@ public class DataNode implements DataNodeMBean {
/* Store cluster ID */
IoTDBDescriptor.getInstance().getConfig().setClusterId(runtimeConfiguration.getClusterId());
+
+ /* Store table info*/
+ DataNodeTableCache.getInstance().init(runtimeConfiguration.getTableInfo());
}
/**
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/SchemaConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/SchemaConstant.java
index d7b40a37d84..7c67375d31d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/SchemaConstant.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/SchemaConstant.java
@@ -87,6 +87,8 @@ public class SchemaConstant {
public static final byte LOGICAL_VIEW_MNODE_TYPE = 5;
+ public static final byte TABLE_MNODE_TYPE = 6;
+
public static final String INTERNAL_MNODE_TYPE_NAME = "InternalMNode";
public static final String STORAGE_GROUP_MNODE_TYPE_NAME = "StorageGroupMNode";
public static final String MEASUREMENT_MNODE_TYPE_NAME = "MeasurementMNode";
@@ -95,6 +97,8 @@ public class SchemaConstant {
public static final String LOGICAL_VIEW_MNODE_TYPE_NAME = "LogicalViewMNode";
+ public static final String TABLE_MNODE_TYPE_NAME = "TableMNode";
+
public static final String SCHEMA_REGION_METRIC_NAME = "schema_region";
public static final String SCHEMA_ENGINE_METRIC_NAME = "pbtree";
@@ -115,6 +119,8 @@ public class SchemaConstant {
return STORAGE_GROUP_ENTITY_MNODE_TYPE_NAME;
case LOGICAL_VIEW_MNODE_TYPE:
return LOGICAL_VIEW_MNODE_TYPE_NAME;
+ case TABLE_MNODE_TYPE:
+ return TABLE_MNODE_TYPE_NAME;
default:
throw new RuntimeException("Undefined MNode type " + type);
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java
new file mode 100644
index 00000000000..6904c000c6d
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTable.java
@@ -0,0 +1,144 @@
+/*
+ * 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.commons.schema.table;
+
+import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchemaUtil;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class TsTable {
+
+ private static final String TIME_COLUMN_NAME = "Time";
+ private static final TimeColumnSchema TIME_COLUMN_SCHEMA =
+ new TimeColumnSchema(TIME_COLUMN_NAME, TSDataType.INT64);
+
+ private final String tableName;
+
+ private final Map<String, TsTableColumnSchema> columnSchemaMap =
+ Collections.synchronizedMap(new LinkedHashMap<>());
+
+ private Map<String, String> props = null;
+
+ public TsTable(String tableName) {
+ this.tableName = tableName;
+ columnSchemaMap.put(TIME_COLUMN_NAME, TIME_COLUMN_SCHEMA);
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public TsTableColumnSchema getColumnSchema(String columnName) {
+ return columnSchemaMap.get(columnName);
+ }
+
+ public void addColumnSchema(TsTableColumnSchema columnSchema) {
+ columnSchemaMap.put(columnSchema.getColumnName(), columnSchema);
+ }
+
+ public int getColumnNum() {
+ return columnSchemaMap.size();
+ }
+
+ public List<TsTableColumnSchema> getColumnList() {
+ return new ArrayList<>(columnSchemaMap.values());
+ }
+
+ public String getPropValue(String propKey) {
+ return props == null ? null : props.get(propKey);
+ }
+
+ public void addProp(String key, String value) {
+ if (props == null) {
+ synchronized (this) {
+ if (props == null) {
+ props = new ConcurrentHashMap<>();
+ }
+ }
+ }
+ props.put(key, value);
+ }
+
+ public byte[] serialize() {
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ try {
+ serialize(stream);
+ } catch (IOException ignored) {
+ // won't happen
+ }
+ return stream.toByteArray();
+ }
+
+ public void serialize(OutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(tableName, stream);
+ ReadWriteIOUtils.write(columnSchemaMap.size(), stream);
+ for (TsTableColumnSchema columnSchema : columnSchemaMap.values()) {
+ TsTableColumnSchemaUtil.serialize(columnSchema, stream);
+ }
+ ReadWriteIOUtils.write(props, stream);
+ }
+
+ public static TsTable deserialize(InputStream inputStream) throws IOException {
+ String name = ReadWriteIOUtils.readString(inputStream);
+ TsTable table = new TsTable(name);
+ int columnNum = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < columnNum; i++) {
+ table.addColumnSchema(TsTableColumnSchemaUtil.deserialize(inputStream));
+ }
+ table.props = ReadWriteIOUtils.readMap(inputStream);
+ return table;
+ }
+
+ public static TsTable deserialize(ByteBuffer buffer) {
+ String name = ReadWriteIOUtils.readString(buffer);
+ TsTable table = new TsTable(name);
+ int columnNum = ReadWriteIOUtils.readInt(buffer);
+ for (int i = 0; i < columnNum; i++) {
+ table.addColumnSchema(TsTableColumnSchemaUtil.deserialize(buffer));
+ }
+ table.props = ReadWriteIOUtils.readMap(buffer);
+ return table;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return super.equals(o);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(tableName);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTableInternalRPCType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTableInternalRPCType.java
new file mode 100644
index 00000000000..c874fc45c88
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTableInternalRPCType.java
@@ -0,0 +1,64 @@
+/*
+ * 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.commons.schema.table;
+
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+public enum TsTableInternalRPCType {
+ PRE_CREATE((byte) 0),
+ ROLLBACK_CREATE((byte) 1),
+ COMMIT_CREATE((byte) 2);
+
+ private final byte operationType;
+
+ private TsTableInternalRPCType(byte operationType) {
+ this.operationType = operationType;
+ }
+
+ public byte getOperationType() {
+ return operationType;
+ }
+
+ public void serialize(OutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(operationType, stream);
+ }
+
+ public static TsTableInternalRPCType deserialize(ByteBuffer buffer) {
+ byte type = ReadWriteIOUtils.readByte(buffer);
+ return getType(type);
+ }
+
+ public static TsTableInternalRPCType getType(byte type) {
+ switch (type) {
+ case 0:
+ return PRE_CREATE;
+ case 1:
+ return ROLLBACK_CREATE;
+ case 2:
+ return COMMIT_CREATE;
+ default:
+ throw new IllegalArgumentException("Unknown table update operation type" + type);
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTableInternalRPCUtil.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTableInternalRPCUtil.java
new file mode 100644
index 00000000000..ce0fc3becb4
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/TsTableInternalRPCUtil.java
@@ -0,0 +1,98 @@
+package org.apache.iotdb.commons.schema.table;
+
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TsTableInternalRPCUtil {
+
+ private TsTableInternalRPCUtil() {
+ // do nothing
+ }
+
+ public static byte[] serializeBatchTsTable(Map<String, List<TsTable>> tableMap) {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try {
+ ReadWriteIOUtils.write(tableMap.size(), outputStream);
+ for (Map.Entry<String, List<TsTable>> entry : tableMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), outputStream);
+ ReadWriteIOUtils.write(entry.getValue().size(), outputStream);
+ for (TsTable table : entry.getValue()) {
+ table.serialize(outputStream);
+ }
+ }
+ } catch (IOException ignored) {
+ // won't happen
+ }
+ return outputStream.toByteArray();
+ }
+
+ public static Map<String, List<TsTable>> deserializeBatchTsTable(byte[] bytes) {
+ InputStream inputStream = new ByteArrayInputStream(bytes);
+ Map<String, List<TsTable>> result = new HashMap<>();
+ try {
+ int dbNum = ReadWriteIOUtils.readInt(inputStream);
+ String database;
+ int tableNum;
+ List<TsTable> tableList;
+ for (int i = 0; i < dbNum; i++) {
+ database = ReadWriteIOUtils.readString(inputStream);
+ tableNum = ReadWriteIOUtils.readInt(inputStream);
+ tableList = new ArrayList<>(tableNum);
+ for (int j = 0; j < tableNum; j++) {
+ tableList.add(TsTable.deserialize(inputStream));
+ }
+ result.put(database, tableList);
+ }
+ } catch (IOException ignored) {
+ // won't happen
+ }
+ return result;
+ }
+
+ public static byte[] serializeSingleTsTable(String database, TsTable table) {
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try {
+ ReadWriteIOUtils.write(database, outputStream);
+ table.serialize(outputStream);
+ } catch (IOException ignored) {
+ // won't happen
+ }
+ return outputStream.toByteArray();
+ }
+
+ public static Pair<String, TsTable> deserializeSingleTsTable(byte[] bytes) {
+ InputStream inputStream = new ByteArrayInputStream(bytes);
+ try {
+ String database = ReadWriteIOUtils.readString(inputStream);
+ TsTable table = TsTable.deserialize(inputStream);
+ return new Pair<>(database, table);
+ } catch (IOException ignored) {
+ // won't happen
+ }
+ throw new IllegalStateException();
+ }
+
+ public static byte[] serializeTableInitializationInfo(
+ Map<String, List<TsTable>> usingTableMap, Map<String, List<TsTable>> preCreateTableMap) {
+ byte[] usingBytes = serializeBatchTsTable(usingTableMap);
+ byte[] preCreateBytes = serializeBatchTsTable(preCreateTableMap);
+ byte[] result = new byte[usingBytes.length + preCreateBytes.length];
+ System.arraycopy(usingBytes, 0, result, 0, usingBytes.length);
+ System.arraycopy(preCreateBytes, 0, result, usingBytes.length, preCreateBytes.length);
+ return result;
+ }
+
+ public static Pair<Map<String, List<TsTable>>, Map<String, List<TsTable>>>
+ deserializeTableInitializationInfo(byte[] bytes) {
+ return new Pair<>(deserializeBatchTsTable(bytes), deserializeBatchTsTable(bytes));
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/AttributeColumnSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/AttributeColumnSchema.java
new file mode 100644
index 00000000000..332bb4efb50
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/AttributeColumnSchema.java
@@ -0,0 +1,57 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class AttributeColumnSchema extends TsTableColumnSchema {
+ public AttributeColumnSchema(String columnName, TSDataType dataType) {
+ super(columnName, dataType);
+ }
+
+ public AttributeColumnSchema(String columnName, TSDataType dataType, Map<String, String> props) {
+ super(columnName, dataType, props);
+ }
+
+ @Override
+ public TsTableColumnCategory getColumnCategory() {
+ return TsTableColumnCategory.ATTRIBUTE;
+ }
+
+ static AttributeColumnSchema deserialize(InputStream stream) throws IOException {
+ String columnName = ReadWriteIOUtils.readString(stream);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(stream);
+ Map<String, String> props = ReadWriteIOUtils.readMap(stream);
+ return new AttributeColumnSchema(columnName, dataType, props);
+ }
+
+ static AttributeColumnSchema deserialize(ByteBuffer buffer) {
+ String columnName = ReadWriteIOUtils.readString(buffer);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(buffer);
+ Map<String, String> props = ReadWriteIOUtils.readMap(buffer);
+ return new AttributeColumnSchema(columnName, dataType, props);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/IdColumnSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/IdColumnSchema.java
new file mode 100644
index 00000000000..2110cfd15ba
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/IdColumnSchema.java
@@ -0,0 +1,57 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class IdColumnSchema extends TsTableColumnSchema {
+ public IdColumnSchema(String columnName, TSDataType dataType) {
+ super(columnName, dataType);
+ }
+
+ public IdColumnSchema(String columnName, TSDataType dataType, Map<String, String> props) {
+ super(columnName, dataType, props);
+ }
+
+ @Override
+ public TsTableColumnCategory getColumnCategory() {
+ return TsTableColumnCategory.ID;
+ }
+
+ static IdColumnSchema deserialize(InputStream stream) throws IOException {
+ String columnName = ReadWriteIOUtils.readString(stream);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(stream);
+ Map<String, String> props = ReadWriteIOUtils.readMap(stream);
+ return new IdColumnSchema(columnName, dataType, props);
+ }
+
+ static IdColumnSchema deserialize(ByteBuffer buffer) {
+ String columnName = ReadWriteIOUtils.readString(buffer);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(buffer);
+ Map<String, String> props = ReadWriteIOUtils.readMap(buffer);
+ return new IdColumnSchema(columnName, dataType, props);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/MeasurementColumnSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/MeasurementColumnSchema.java
new file mode 100644
index 00000000000..554396ec395
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/MeasurementColumnSchema.java
@@ -0,0 +1,101 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class MeasurementColumnSchema extends TsTableColumnSchema {
+
+ private final TSEncoding encoding;
+
+ private final CompressionType compressor;
+
+ public MeasurementColumnSchema(
+ String columnName, TSDataType dataType, TSEncoding encoding, CompressionType compressor) {
+ super(columnName, dataType);
+ this.encoding = encoding;
+ this.compressor = compressor;
+ }
+
+ public MeasurementColumnSchema(
+ String columnName,
+ TSDataType dataType,
+ TSEncoding encoding,
+ CompressionType compressor,
+ Map<String, String> props) {
+ super(columnName, dataType, props);
+ this.encoding = encoding;
+ this.compressor = compressor;
+ }
+
+ @Override
+ public TsTableColumnCategory getColumnCategory() {
+ return TsTableColumnCategory.MEASUREMENT;
+ }
+
+ public TSEncoding getEncoding() {
+ return encoding;
+ }
+
+ public CompressionType getCompressor() {
+ return compressor;
+ }
+
+ public MeasurementSchema getMeasurementSchema() {
+ return new MeasurementSchema(columnName, dataType, encoding, compressor, props);
+ }
+
+ @Override
+ void serialize(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(columnName, outputStream);
+ ReadWriteIOUtils.write(dataType, outputStream);
+ ReadWriteIOUtils.write(encoding, outputStream);
+ ReadWriteIOUtils.write(compressor, outputStream);
+ ReadWriteIOUtils.write(props, outputStream);
+ }
+
+ static MeasurementColumnSchema deserialize(InputStream stream) throws IOException {
+ String columnName = ReadWriteIOUtils.readString(stream);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(stream);
+ TSEncoding encoding = ReadWriteIOUtils.readEncoding(stream);
+ CompressionType compressor = ReadWriteIOUtils.readCompressionType(stream);
+ Map<String, String> props = ReadWriteIOUtils.readMap(stream);
+ return new MeasurementColumnSchema(columnName, dataType, encoding, compressor, props);
+ }
+
+ static MeasurementColumnSchema deserialize(ByteBuffer buffer) {
+ String columnName = ReadWriteIOUtils.readString(buffer);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(buffer);
+ TSEncoding encoding = ReadWriteIOUtils.readEncoding(buffer);
+ CompressionType compressor = ReadWriteIOUtils.readCompressionType(buffer);
+ Map<String, String> props = ReadWriteIOUtils.readMap(buffer);
+ return new MeasurementColumnSchema(columnName, dataType, encoding, compressor, props);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TimeColumnSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TimeColumnSchema.java
new file mode 100644
index 00000000000..2aa0a62797c
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TimeColumnSchema.java
@@ -0,0 +1,57 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class TimeColumnSchema extends TsTableColumnSchema {
+ public TimeColumnSchema(String columnName, TSDataType dataType) {
+ super(columnName, dataType);
+ }
+
+ public TimeColumnSchema(String columnName, TSDataType dataType, Map<String, String> props) {
+ super(columnName, dataType, props);
+ }
+
+ @Override
+ public TsTableColumnCategory getColumnCategory() {
+ return TsTableColumnCategory.TIME;
+ }
+
+ static TimeColumnSchema deserialize(InputStream stream) throws IOException {
+ String columnName = ReadWriteIOUtils.readString(stream);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(stream);
+ Map<String, String> props = ReadWriteIOUtils.readMap(stream);
+ return new TimeColumnSchema(columnName, dataType, props);
+ }
+
+ static TimeColumnSchema deserialize(ByteBuffer buffer) {
+ String columnName = ReadWriteIOUtils.readString(buffer);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(buffer);
+ Map<String, String> props = ReadWriteIOUtils.readMap(buffer);
+ return new TimeColumnSchema(columnName, dataType, props);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java
new file mode 100644
index 00000000000..074916c2564
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnCategory.java
@@ -0,0 +1,73 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+public enum TsTableColumnCategory {
+ ID((byte) 0),
+ ATTRIBUTE((byte) 1),
+ TIME((byte) 2),
+ MEASUREMENT((byte) 3);
+
+ private final byte category;
+
+ private TsTableColumnCategory(byte category) {
+ this.category = category;
+ }
+
+ byte getValue() {
+ return category;
+ }
+
+ public void serialize(OutputStream stream) throws IOException {
+ ReadWriteIOUtils.write(category, stream);
+ }
+
+ public static TsTableColumnCategory deserialize(InputStream stream) throws IOException {
+ byte category = (byte) stream.read();
+ return deserialize(category);
+ }
+
+ public static TsTableColumnCategory deserialize(ByteBuffer stream) {
+ byte category = stream.get();
+ return deserialize(category);
+ }
+
+ private static TsTableColumnCategory deserialize(byte category) {
+ switch (category) {
+ case 0:
+ return ID;
+ case 1:
+ return ATTRIBUTE;
+ case 2:
+ return TIME;
+ case 3:
+ return MEASUREMENT;
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnSchema.java
new file mode 100644
index 00000000000..73f475ecade
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnSchema.java
@@ -0,0 +1,74 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Objects;
+
+public abstract class TsTableColumnSchema {
+
+ protected String columnName;
+
+ protected TSDataType dataType;
+
+ protected Map<String, String> props = null;
+
+ TsTableColumnSchema(String columnName, TSDataType dataType) {
+ this.columnName = columnName;
+ this.dataType = dataType;
+ }
+
+ TsTableColumnSchema(String columnName, TSDataType dataType, Map<String, String> props) {
+ this.columnName = columnName;
+ this.dataType = dataType;
+ this.props = props;
+ }
+
+ public String getColumnName() {
+ return columnName;
+ }
+
+ public TSDataType getDataType() {
+ return dataType;
+ }
+
+ public abstract TsTableColumnCategory getColumnCategory();
+
+ void serialize(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(columnName, outputStream);
+ ReadWriteIOUtils.write(dataType, outputStream);
+ ReadWriteIOUtils.write(props, outputStream);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return super.equals(o);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(columnName);
+ }
+}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnSchemaUtil.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnSchemaUtil.java
new file mode 100644
index 00000000000..5cd052b2f3f
--- /dev/null
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/column/TsTableColumnSchemaUtil.java
@@ -0,0 +1,80 @@
+/*
+ * 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.commons.schema.table.column;
+
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+public class TsTableColumnSchemaUtil {
+
+ private TsTableColumnSchemaUtil() {
+ // do nothing
+ }
+
+ public static void serialize(TsTableColumnSchema columnSchema, OutputStream outputStream)
+ throws IOException {
+ ReadWriteIOUtils.write(columnSchema.getColumnCategory().getValue(), outputStream);
+ columnSchema.serialize(outputStream);
+ }
+
+ public static TsTableColumnSchema deserialize(InputStream inputStream) throws IOException {
+ return deserialize(TsTableColumnCategory.deserialize(inputStream), inputStream);
+ }
+
+ private static TsTableColumnSchema deserialize(TsTableColumnCategory category, InputStream stream)
+ throws IOException {
+ switch (category) {
+ case ID:
+ return IdColumnSchema.deserialize(stream);
+ case ATTRIBUTE:
+ return AttributeColumnSchema.deserialize(stream);
+ case TIME:
+ return TimeColumnSchema.deserialize(stream);
+ case MEASUREMENT:
+ return MeasurementColumnSchema.deserialize(stream);
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+
+ public static TsTableColumnSchema deserialize(ByteBuffer buffer) {
+ return deserialize(TsTableColumnCategory.deserialize(buffer), buffer);
+ }
+
+ private static TsTableColumnSchema deserialize(
+ TsTableColumnCategory category, ByteBuffer buffer) {
+ switch (category) {
+ case ID:
+ return IdColumnSchema.deserialize(buffer);
+ case ATTRIBUTE:
+ return AttributeColumnSchema.deserialize(buffer);
+ case TIME:
+ return TimeColumnSchema.deserialize(buffer);
+ case MEASUREMENT:
+ return MeasurementColumnSchema.deserialize(buffer);
+ default:
+ throw new IllegalArgumentException();
+ }
+ }
+}
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 28978144ae8..b14f67cef03 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -105,6 +105,7 @@ struct TRuntimeConfiguration {
4: required binary allTTLInformation
5: required list<binary> allPipeInformation
6: optional string clusterId
+ 7: optional binary tableInfo
}
struct TDataNodeRegisterReq {
@@ -897,6 +898,7 @@ struct TShowThrottleReq {
1: optional string userName;
}
+
// ====================================================
// Activation
// ====================================================
@@ -1557,5 +1559,11 @@ service IConfigNodeRPCService {
/** Get throttle quota information */
TThrottleQuotaResp getThrottleQuota()
+
+ // ======================================================
+ // Table
+ // ======================================================
+
+ common.TSStatus createTable(binary tableInfo)
}
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 5e9987d2a50..838585cbf10 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -325,6 +325,11 @@ struct TUpdateTemplateReq {
2: required binary templateInfo
}
+struct TUpdateTableReq {
+ 1: required byte type
+ 2: required binary tableInfo
+}
+
struct TTsFilePieceReq {
1: required binary body
2: required string uuid
@@ -974,6 +979,11 @@ service IDataNodeRPCService {
* Fetch fragment instance statistics for EXPLAIN ANALYZE
*/
TFetchFragmentInstanceStatisticsResp fetchFragmentInstanceStatistics(TFetchFragmentInstanceStatisticsReq req)
+
+ /**
+ * Update Table Cache
+ */
+ common.TSStatus updateTable(TUpdateTableReq req)
}
service MPPDataExchangeService {