Implement alter database in table model

apache · Dec 6, 2024 · 5c84e68 · 5c84e68
1 parent a9c5547
commit 5c84e68
Show file tree

Hide file tree

Showing 28 changed files with 457 additions and 144 deletions.
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableIT.java
@@ -299,10 +299,15 @@ public void testManageTable() {
         assertEquals(tableNames.length, cnt);
       }
 
+      // Will not affect the manual "6600000"
+      statement.execute("alter database test2 set properties ttl=6600000");
+      statement.execute("alter database test2 set properties ttl=DEFAULT");
+
       statement.execute("alter table table3 set properties ttl=1000000");
       statement.execute("alter table table3 set properties ttl=DEFAULT");
 
-      // The table3's ttl shall be also 3000000
+      ttls = new String[] {"INF", "6600000"};
+      // The table3's ttl shall be "INF"
       try (final ResultSet resultSet = statement.executeQuery("SHOW tables")) {
         int cnt = 0;
         ResultSetMetaData metaData = resultSet.getMetaData();

diff --git a/...ignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/...ignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -207,7 +207,7 @@ public TSStatus setDatabase(
   public TSStatus alterDatabase(
       final DatabaseSchemaPlan databaseSchemaPlan, final boolean isGeneratedByPipe) {
     TSStatus result;
-    TDatabaseSchema databaseSchema = databaseSchemaPlan.getSchema();
+    final TDatabaseSchema databaseSchema = databaseSchemaPlan.getSchema();
 
     if (!isDatabaseExist(databaseSchema.getName())) {
       // Reject if Database doesn't exist
@@ -219,7 +219,7 @@ public TSStatus alterDatabase(
 
     if (databaseSchema.isSetMinSchemaRegionGroupNum()) {
       // Validate alter SchemaRegionGroupNum
-      int minSchemaRegionGroupNum =
+      final int minSchemaRegionGroupNum =
           getMinRegionGroupNum(databaseSchema.getName(), TConsensusGroupType.SchemaRegion);
       if (databaseSchema.getMinSchemaRegionGroupNum() <= minSchemaRegionGroupNum) {
         result = new TSStatus(TSStatusCode.DATABASE_CONFIG_ERROR.getStatusCode());

diff --git a/...gnode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java b/...gnode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ClusterSchemaInfo.java
@@ -246,10 +246,19 @@ public TSStatus alterDatabase(final DatabaseSchemaPlan plan) {
             currentSchema.getMaxDataRegionGroupNum());
       }
 
+      if (alterSchema.isSetTTL()) {
+        currentSchema.setTTL(alterSchema.getTTL());
+        LOGGER.info(
+            "[SetTTL] The ttl of Database: {} is adjusted to: {}",
+            currentSchema.getName(),
+            currentSchema.getTTL());
+      }
+
       mTree
           .getDatabaseNodeByDatabasePath(partialPathName)
           .getAsMNode()
           .setDatabaseSchema(currentSchema);
+
       result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
     } catch (final MetadataException e) {
       LOGGER.error(ERROR_NAME, e);

diff --git a/.../confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/.../confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
@@ -827,7 +827,8 @@ public void setTableProperties(
           if (Objects.nonNull(v)) {
             table.addProp(k, v);
           } else if (k.equals(TsTable.TTL_PROPERTY)
-              && databaseNode.getDatabaseSchema().isSetTTL()) {
+              && databaseNode.getDatabaseSchema().isSetTTL()
+              && databaseNode.getDatabaseSchema().getTTL() != Long.MAX_VALUE) {
             table.addProp(k, String.valueOf(databaseNode.getDatabaseSchema().getTTL()));
           } else {
             table.removeProp(k);

diff --git a/.../main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java b/.../main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java
@@ -65,11 +65,12 @@ public class DeleteDatabaseProcedure
 
   private TDatabaseSchema deleteDatabaseSchema;
 
-  public DeleteDatabaseProcedure(boolean isGeneratedByPipe) {
+  public DeleteDatabaseProcedure(final boolean isGeneratedByPipe) {
     super(isGeneratedByPipe);
   }
 
-  public DeleteDatabaseProcedure(TDatabaseSchema deleteDatabaseSchema, boolean isGeneratedByPipe) {
+  public DeleteDatabaseProcedure(
+      final TDatabaseSchema deleteDatabaseSchema, final boolean isGeneratedByPipe) {
     super(isGeneratedByPipe);
     this.deleteDatabaseSchema = deleteDatabaseSchema;
   }
@@ -78,12 +79,13 @@ public TDatabaseSchema getDeleteDatabaseSchema() {
     return deleteDatabaseSchema;
   }
 
-  public void setDeleteDatabaseSchema(TDatabaseSchema deleteDatabaseSchema) {
+  public void setDeleteDatabaseSchema(final TDatabaseSchema deleteDatabaseSchema) {
     this.deleteDatabaseSchema = deleteDatabaseSchema;
   }
 
   @Override
-  protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupState state)
+  protected Flow executeFromState(
+      final ConfigNodeProcedureEnv env, final DeleteStorageGroupState state)
       throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
     if (deleteDatabaseSchema == null) {
       return Flow.NO_MORE_STATE;
@@ -113,11 +115,11 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
               deleteDatabaseSchema.getName());
 
           // Submit RegionDeleteTasks
-          OfferRegionMaintainTasksPlan dataRegionDeleteTaskOfferPlan =
+          final OfferRegionMaintainTasksPlan dataRegionDeleteTaskOfferPlan =
               new OfferRegionMaintainTasksPlan();
-          List<TRegionReplicaSet> regionReplicaSets =
+          final List<TRegionReplicaSet> regionReplicaSets =
               env.getAllReplicaSets(deleteDatabaseSchema.getName());
-          List<TRegionReplicaSet> schemaRegionReplicaSets = new ArrayList<>();
+          final List<TRegionReplicaSet> schemaRegionReplicaSets = new ArrayList<>();
           regionReplicaSets.forEach(
               regionReplicaSet -> {
                 // Clear heartbeat cache along the way
@@ -147,12 +149,12 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
           }
 
           // try sync delete schemaengine region
-          DataNodeAsyncRequestContext<TConsensusGroupId, TSStatus> asyncClientHandler =
+          final DataNodeAsyncRequestContext<TConsensusGroupId, TSStatus> asyncClientHandler =
               new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.DELETE_REGION);
-          Map<Integer, RegionDeleteTask> schemaRegionDeleteTaskMap = new HashMap<>();
+          final Map<Integer, RegionDeleteTask> schemaRegionDeleteTaskMap = new HashMap<>();
           int requestIndex = 0;
-          for (TRegionReplicaSet schemaRegionReplicaSet : schemaRegionReplicaSets) {
-            for (TDataNodeLocation dataNodeLocation :
+          for (final TRegionReplicaSet schemaRegionReplicaSet : schemaRegionReplicaSets) {
+            for (final TDataNodeLocation dataNodeLocation :
                 schemaRegionReplicaSet.getDataNodeLocations()) {
               asyncClientHandler.putRequest(requestIndex, schemaRegionReplicaSet.getRegionId());
               asyncClientHandler.putNodeLocation(requestIndex, dataNodeLocation);
@@ -165,7 +167,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
           if (!schemaRegionDeleteTaskMap.isEmpty()) {
             CnToDnInternalServiceAsyncRequestManager.getInstance()
                 .sendAsyncRequestWithRetry(asyncClientHandler);
-            for (Map.Entry<Integer, TSStatus> entry :
+            for (final Map.Entry<Integer, TSStatus> entry :
                 asyncClientHandler.getResponseMap().entrySet()) {
               if (entry.getValue().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
                 LOG.info(
@@ -183,7 +185,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
 
             if (!schemaRegionDeleteTaskMap.isEmpty()) {
               // submit async schemaengine region delete task for failed sync execution
-              OfferRegionMaintainTasksPlan schemaRegionDeleteTaskOfferPlan =
+              final OfferRegionMaintainTasksPlan schemaRegionDeleteTaskOfferPlan =
                   new OfferRegionMaintainTasksPlan();
               schemaRegionDeleteTaskMap
                   .values()
@@ -215,7 +217,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
                 new ProcedureException("[DeleteDatabaseProcedure] Delete DatabaseSchema failed"));
           }
       }
-    } catch (ConsensusException | TException | IOException e) {
+    } catch (final ConsensusException | TException | IOException e) {
       if (isRollbackSupported(state)) {
         setFailure(
             new ProcedureException(
@@ -238,7 +240,8 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, DeleteStorageGroupSt
   }
 
   @Override
-  protected void rollbackState(ConfigNodeProcedureEnv env, DeleteStorageGroupState state)
+  protected void rollbackState(
+      final ConfigNodeProcedureEnv env, final DeleteStorageGroupState state)
       throws IOException, InterruptedException {
     switch (state) {
       case PRE_DELETE_DATABASE:
@@ -254,7 +257,7 @@ protected void rollbackState(ConfigNodeProcedureEnv env, DeleteStorageGroupState
   }
 
   @Override
-  protected boolean isRollbackSupported(DeleteStorageGroupState state) {
+  protected boolean isRollbackSupported(final DeleteStorageGroupState state) {
     switch (state) {
       case PRE_DELETE_DATABASE:
       case INVALIDATE_CACHE:
@@ -265,12 +268,12 @@ protected boolean isRollbackSupported(DeleteStorageGroupState state) {
   }
 
   @Override
-  protected DeleteStorageGroupState getState(int stateId) {
+  protected DeleteStorageGroupState getState(final int stateId) {
     return DeleteStorageGroupState.values()[stateId];
   }
 
   @Override
-  protected int getStateId(DeleteStorageGroupState deleteStorageGroupState) {
+  protected int getStateId(final DeleteStorageGroupState deleteStorageGroupState) {
     return deleteStorageGroupState.ordinal();
   }
 
@@ -284,7 +287,7 @@ public String getDatabase() {
   }
 
   @Override
-  public void serialize(DataOutputStream stream) throws IOException {
+  public void serialize(final DataOutputStream stream) throws IOException {
     stream.writeShort(
         isGeneratedByPipe
             ? ProcedureType.PIPE_ENRICHED_DELETE_DATABASE_PROCEDURE.getTypeCode()
@@ -294,17 +297,17 @@ public void serialize(DataOutputStream stream) throws IOException {
   }
 
   @Override
-  public void deserialize(ByteBuffer byteBuffer) {
+  public void deserialize(final ByteBuffer byteBuffer) {
     super.deserialize(byteBuffer);
     try {
       deleteDatabaseSchema = ThriftConfigNodeSerDeUtils.deserializeTDatabaseSchema(byteBuffer);
-    } catch (ThriftSerDeException e) {
-      LOG.error("Error in deserialize DeleteStorageGroupProcedure", e);
+    } catch (final ThriftSerDeException e) {
+      LOG.error("Error in deserialize DeleteDatabaseProcedure", e);
     }
   }
 
   @Override
-  public boolean equals(Object that) {
+  public boolean equals(final Object that) {
     if (that instanceof DeleteDatabaseProcedure) {
       DeleteDatabaseProcedure thatProc = (DeleteDatabaseProcedure) that;
       return thatProc.getProcId() == this.getProcId()

diff --git a/...nfignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java b/...nfignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java
@@ -225,7 +225,7 @@ protected void onAllReplicasetFailure(
   public static Map<Integer, TSStatus> preReleaseTable(
       final String database, final TsTable table, final ConfigManager configManager) {
     final TUpdateTableReq req = new TUpdateTableReq();
-    req.setType(TsTableInternalRPCType.PRE_CREATE_OR_ADD_COLUMN.getOperationType());
+    req.setType(TsTableInternalRPCType.PRE_UPDATE_TABLE.getOperationType());
     req.setTableInfo(TsTableInternalRPCUtil.serializeSingleTsTableWithDatabase(database, table));
 
     final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
@@ -242,7 +242,7 @@ public static Map<Integer, TSStatus> preReleaseTable(
   public static Map<Integer, TSStatus> commitReleaseTable(
       final String database, final String tableName, final ConfigManager configManager) {
     final TUpdateTableReq req = new TUpdateTableReq();
-    req.setType(TsTableInternalRPCType.COMMIT_CREATE_OR_ADD_COLUMN.getOperationType());
+    req.setType(TsTableInternalRPCType.COMMIT_UPDATE_TABLE.getOperationType());
     final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
     try {
       ReadWriteIOUtils.write(database, outputStream);
@@ -266,7 +266,7 @@ public static Map<Integer, TSStatus> commitReleaseTable(
   public static Map<Integer, TSStatus> rollbackPreRelease(
       final String database, final String tableName, final ConfigManager configManager) {
     final TUpdateTableReq req = new TUpdateTableReq();
-    req.setType(TsTableInternalRPCType.ROLLBACK_CREATE_OR_ADD_COLUMN.getOperationType());
+    req.setType(TsTableInternalRPCType.ROLLBACK_UPDATE_TABLE.getOperationType());
     final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
     try {
       ReadWriteIOUtils.write(database, outputStream);

diff --git a/...in/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java b/...in/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedure.java
@@ -145,7 +145,9 @@ private void checkTableExistence(final ConfigNodeProcedureEnv env) {
       } else {
         final TDatabaseSchema schema =
             env.getConfigManager().getClusterSchemaManager().getDatabaseSchemaByName(database);
-        if (schema.isSetTTL() && !table.getPropValue(TsTable.TTL_PROPERTY).isPresent()) {
+        if (!table.getPropValue(TsTable.TTL_PROPERTY).isPresent()
+            && schema.isSetTTL()
+            && schema.getTTL() != Long.MAX_VALUE) {
           table.addProp(TsTable.TTL_PROPERTY, String.valueOf(schema.getTTL()));
         }
         setNextState(CreateTableState.PRE_CREATE);

diff --git a/.../src/main/java/org/apache/iotdb/confignode/procedure/state/schema/AlterDatabaseState.java b/.../src/main/java/org/apache/iotdb/confignode/procedure/state/schema/AlterDatabaseState.java
@@ -0,0 +1,27 @@
+/*
+ * 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 AlterDatabaseState {
+  CHECK_ALTERED_TABLES,
+  PRE_RELEASE,
+  ALTER_DATABASE,
+  COMMIT_RELEASE
+}
diff --git a/...onfignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java b/...onfignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
@@ -320,7 +320,7 @@ public Procedure create(ByteBuffer buffer) throws IOException {
     return procedure;
   }
 
-  public static ProcedureType getProcedureType(Procedure<?> procedure) {
+  public static ProcedureType getProcedureType(final Procedure<?> procedure) {
     if (procedure instanceof DeleteDatabaseProcedure) {
       return ProcedureType.DELETE_DATABASE_PROCEDURE;
     } else if (procedure instanceof AddConfigNodeProcedure) {

diff --git a/...c/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/...c/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -481,7 +481,7 @@ public TSStatus alterDatabase(final TDatabaseSchema databaseSchema) {
     TSStatus errorResp = null;
 
     // TODO: Support alter the following fields
-    if (databaseSchema.isSetTTL()) {
+    if (databaseSchema.isSetTTL() && !databaseSchema.isIsTableModel()) {
       errorResp =
           new TSStatus(TSStatusCode.DATABASE_CONFIG_ERROR.getStatusCode())
               .setMessage("Failed to alter database. Doesn't support ALTER TTL yet.");

diff --git a/...rc/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/...rc/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -1533,8 +1533,10 @@ public TFetchFragmentInstanceStatisticsResp fetchFragmentInstanceStatistics(
 
   @Override
   public TSStatus updateTable(final TUpdateTableReq req) {
+    final String database;
+    final int size;
     switch (TsTableInternalRPCType.getType(req.type)) {
-      case PRE_CREATE_OR_ADD_COLUMN:
+      case PRE_UPDATE_TABLE:
         DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.TIMESERIES_VS_TABLE);
         try {
           Pair<String, TsTable> pair =
@@ -1545,13 +1547,13 @@ public TSStatus updateTable(final TUpdateTableReq req) {
               .releaseWriteLock(SchemaLockType.TIMESERIES_VS_TABLE);
         }
         break;
-      case ROLLBACK_CREATE_OR_ADD_COLUMN:
+      case ROLLBACK_UPDATE_TABLE:
         DataNodeTableCache.getInstance()
             .rollbackUpdateTable(
                 ReadWriteIOUtils.readString(req.tableInfo),
                 ReadWriteIOUtils.readString(req.tableInfo));
         break;
-      case COMMIT_CREATE_OR_ADD_COLUMN:
+      case COMMIT_UPDATE_TABLE:
         DataNodeTableCache.getInstance()
             .commitUpdateTable(
                 ReadWriteIOUtils.readString(req.tableInfo),

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
@@ -56,6 +56,7 @@
 import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
 import org.apache.iotdb.db.queryengine.plan.relational.security.AllowAllAccessControl;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AddColumn;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ClearCache;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateDB;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CreateTable;
@@ -369,6 +370,7 @@ private IQueryExecution createQueryExecutionForTableModel(
     if (statement instanceof DropDB
         || statement instanceof ShowDB
         || statement instanceof CreateDB
+        || statement instanceof AlterDB
         || statement instanceof Use
         || statement instanceof CreateTable
         || statement instanceof DescribeTable