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/07/10 03:52:06 UTC
(iotdb) branch ty/TableModelGrammar updated: Fixed the bug that add column procedure is not idempotent & Added tests for table plan / procedures
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 0526599c961 Fixed the bug that add column procedure is not idempotent & Added tests for table plan / procedures
0526599c961 is described below
commit 0526599c9615c37e1acbbe548255bf0cd225957f
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Jul 10 11:51:59 2024 +0800
Fixed the bug that add column procedure is not idempotent & Added tests for table plan / procedures
---
.../manager/schema/ClusterSchemaManager.java | 45 ++---
.../impl/schema/table/AddTableColumnProcedure.java | 201 +++++++++++----------
.../impl/schema/table/CreateTableProcedure.java | 134 +++++++-------
.../request/ConfigPhysicalPlanSerDeTest.java | 76 ++++++++
.../schema/table/AddTableColumnProcedureTest.java | 62 +++++++
.../schema/table/CreateTableProcedureTest.java | 73 ++++++++
.../impl/DataNodeInternalRPCServiceImpl.java | 36 +---
.../db/schemaengine/table/DataNodeTableCache.java | 140 ++++----------
.../iotdb/db/schemaengine/table/ITableCache.java | 22 +--
.../schema/table/TsTableInternalRPCType.java | 30 +--
.../schema/table/TsTableInternalRPCUtil.java | 12 +-
.../table/column/TsTableColumnSchemaUtil.java | 4 +-
12 files changed, 479 insertions(+), 356 deletions(-)
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 0cb3bed3639..b016b59d5bb 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
@@ -99,6 +99,7 @@ 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 java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
@@ -1108,18 +1109,17 @@ public class ClusterSchemaManager {
return clusterSchemaInfo.getTsTable(database, tableName);
}
- public synchronized Pair<TSStatus, List<TsTableColumnSchema>> tableColumnCheckForColumnExtension(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList) {
- Map<String, List<TsTable>> currentUsingTable = clusterSchemaInfo.getAllUsingTables();
- TsTable targetTable = null;
- for (TsTable table : currentUsingTable.get(database)) {
- if (table.getTableName().equals(tableName)) {
- targetTable = table;
- break;
- }
- }
-
- if (targetTable == null) {
+ public synchronized Pair<TSStatus, TsTable> tableColumnCheckForColumnExtension(
+ final String database,
+ final String tableName,
+ final List<TsTableColumnSchema> columnSchemaList) {
+ final TsTable originalTable =
+ clusterSchemaInfo.getAllUsingTables().get(database).stream()
+ .filter(tsTable -> tsTable.getTableName().equals(tableName))
+ .findAny()
+ .orElse(null);
+
+ if (Objects.isNull(originalTable)) {
return new Pair<>(
RpcUtils.getStatus(
TSStatusCode.TABLE_NOT_EXISTS,
@@ -1127,15 +1127,18 @@ public class ClusterSchemaManager {
null);
}
- List<TsTableColumnSchema> copiedList = new ArrayList<>();
- for (TsTableColumnSchema columnSchema : columnSchemaList) {
- TsTableColumnSchema existingColumnSchema =
- targetTable.getColumnSchema(columnSchema.getColumnName());
- if (existingColumnSchema == null) {
- copiedList.add(columnSchema);
- }
- }
- return new Pair<>(RpcUtils.SUCCESS_STATUS, copiedList);
+ final TsTable expandedTable = TsTable.deserialize(ByteBuffer.wrap(originalTable.serialize()));
+
+ columnSchemaList.removeIf(
+ columnSchema -> {
+ if (Objects.isNull(originalTable.getColumnSchema(columnSchema.getColumnName()))) {
+ expandedTable.addColumnSchema(columnSchema);
+ return false;
+ }
+ return true;
+ });
+
+ return new Pair<>(RpcUtils.SUCCESS_STATUS, expandedTable);
}
public synchronized TSStatus addTableColumn(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java
index 45e03b1c078..b6017c1ce25 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedure.java
@@ -23,7 +23,9 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.MetadataException;
+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.table.column.TsTableColumnSchema;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchemaUtil;
import org.apache.iotdb.confignode.client.CnToDnRequestType;
@@ -63,27 +65,26 @@ public class AddTableColumnProcedure
private String queryId;
- private List<TsTableColumnSchema> inputColumnList;
-
- private List<TsTableColumnSchema> actualAddedColumnList;
+ private List<TsTableColumnSchema> addedColumnList;
+ private TsTable table;
public AddTableColumnProcedure() {}
public AddTableColumnProcedure(
- String database,
- String tableName,
- String queryId,
- List<TsTableColumnSchema> inputColumnList) {
+ final String database,
+ final String tableName,
+ final String queryId,
+ final List<TsTableColumnSchema> addedColumnList) {
this.database = database;
this.tableName = tableName;
this.queryId = queryId;
- this.inputColumnList = inputColumnList;
+ this.addedColumnList = addedColumnList;
}
@Override
- protected Flow executeFromState(ConfigNodeProcedureEnv env, AddTableColumnState state)
+ protected Flow executeFromState(final ConfigNodeProcedureEnv env, final AddTableColumnState state)
throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
- long startTime = System.currentTimeMillis();
+ final long startTime = System.currentTimeMillis();
try {
switch (state) {
case COLUMN_CHECK:
@@ -118,74 +119,85 @@ public class AddTableColumnProcedure
}
}
- private void columnCheck(ConfigNodeProcedureEnv env) {
- Pair<TSStatus, List<TsTableColumnSchema>> result =
+ private void columnCheck(final ConfigNodeProcedureEnv env) {
+ final Pair<TSStatus, TsTable> result =
env.getConfigManager()
.getClusterSchemaManager()
- .tableColumnCheckForColumnExtension(database, tableName, inputColumnList);
- TSStatus status = result.getLeft();
+ .tableColumnCheckForColumnExtension(database, tableName, addedColumnList);
+ final TSStatus status = result.getLeft();
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())));
return;
}
- actualAddedColumnList = result.getRight();
+ table = result.getRight();
setNextState(AddTableColumnState.PRE_RELEASE);
}
- private void preRelease(ConfigNodeProcedureEnv env) {
- Map<Integer, TDataNodeLocation> dataNodeLocationMap =
- env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
-
- TUpdateTableReq req =
- new TUpdateTableReq(
- TsTableInternalRPCType.PRE_ADD_COLUMN.getOperationType(), getCacheRequestInfo());
+ private void preRelease(final ConfigNodeProcedureEnv env) {
+ final TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.PRE_CREATE_OR_ADD_COLUMN.getOperationType());
+ req.setTableInfo(TsTableInternalRPCUtil.serializeSingleTsTable(database, table));
- DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (TSStatus status : statusMap.values()) {
- // all dataNodes must clear the related schema cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ // All dataNodes must clear the related schema cache
LOGGER.warn(
- "Failed to pre-release column extension info of table {}.{}", database, tableName);
+ "Failed to pre-release column extension info of table {}.{} to DataNode {}",
+ database,
+ table.getTableName(),
+ dataNodeLocationMap.get(entry.getKey()));
setFailure(
new ProcedureException(
new MetadataException("Pre-release table column extension info failed")));
return;
}
}
+
setNextState(AddTableColumnState.ADD_COLUMN);
}
- private void commitRelease(ConfigNodeProcedureEnv env) {
- Map<Integer, TDataNodeLocation> dataNodeLocationMap =
- env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
-
- TUpdateTableReq req =
- new TUpdateTableReq(
- TsTableInternalRPCType.COMMIT_ADD_COLUMN.getOperationType(), getCacheRequestInfo());
+ private void commitRelease(final ConfigNodeProcedureEnv env) {
+ final TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.COMMIT_CREATE_OR_ADD_COLUMN.getOperationType());
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try {
+ ReadWriteIOUtils.write(database, outputStream);
+ ReadWriteIOUtils.write(table.getTableName(), outputStream);
+ } catch (final IOException ignored) {
+ // ByteArrayOutputStream will not throw IOException
+ }
+ req.setTableInfo(outputStream.toByteArray());
- DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (TSStatus status : statusMap.values()) {
- // all dataNodes must clear the related schema cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn("Failed to commit column extension info of table {}.{}", database, tableName);
- // todo async retry until success
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ "Failed to commit column extension info of table {}.{} to DataNode {}",
+ database,
+ table.getTableName(),
+ dataNodeLocationMap.get(entry.getKey()));
+ // TODO: Handle commit failure
return;
}
}
- setNextState(AddTableColumnState.ADD_COLUMN);
}
- private void addColumn(ConfigNodeProcedureEnv env) {
- TSStatus status =
+ private void addColumn(final ConfigNodeProcedureEnv env) {
+ final TSStatus status =
env.getConfigManager()
.getClusterSchemaManager()
- .addTableColumn(database, tableName, inputColumnList);
+ .addTableColumn(database, tableName, addedColumnList);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())));
} else {
@@ -194,16 +206,16 @@ public class AddTableColumnProcedure
}
@Override
- protected void rollbackState(ConfigNodeProcedureEnv env, AddTableColumnState state)
+ protected void rollbackState(final ConfigNodeProcedureEnv env, final AddTableColumnState state)
throws IOException, InterruptedException, ProcedureException {
- long startTime = System.currentTimeMillis();
+ final long startTime = System.currentTimeMillis();
try {
switch (state) {
case ADD_COLUMN:
rollbackAddColumn(env);
break;
case PRE_RELEASE:
- rollbackUpdateCache(env);
+ rollbackPreRelease(env);
break;
}
} finally {
@@ -212,61 +224,59 @@ public class AddTableColumnProcedure
}
}
- private void rollbackAddColumn(ConfigNodeProcedureEnv env) {
- if (actualAddedColumnList == null) {
+ private void rollbackAddColumn(final ConfigNodeProcedureEnv env) {
+ if (table == null) {
return;
}
- TSStatus status =
+ final TSStatus status =
env.getConfigManager()
.getClusterSchemaManager()
- .rollbackAddTableColumn(database, tableName, actualAddedColumnList);
+ .rollbackAddTableColumn(database, tableName, addedColumnList);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
setFailure(new ProcedureException(new IoTDBException(status.getMessage(), status.getCode())));
}
}
- private void rollbackUpdateCache(ConfigNodeProcedureEnv env) {
- Map<Integer, TDataNodeLocation> dataNodeLocationMap =
- env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
-
- TUpdateTableReq req =
- new TUpdateTableReq(
- TsTableInternalRPCType.ROLLBACK_ADD_COLUMN.getOperationType(), getCacheRequestInfo());
+ private void rollbackPreRelease(final ConfigNodeProcedureEnv env) {
+ final TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.ROLLBACK_CREATE_OR_ADD_COLUMN.getOperationType());
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ try {
+ ReadWriteIOUtils.write(database, outputStream);
+ ReadWriteIOUtils.write(table.getTableName(), outputStream);
+ } catch (final IOException ignore) {
+ // ByteArrayOutputStream will not throw IOException
+ }
+ req.setTableInfo(outputStream.toByteArray());
- DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
+ final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (TSStatus status : statusMap.values()) {
- // all dataNodes must clear the related schema cache
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn("Failed to rollback cache of table {}.{}", database, tableName);
- setFailure(new ProcedureException(new MetadataException("Rollback table cache failed")));
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ "Failed to rollback pre-release column extension info of table {}.{} rollback-create info to DataNode {}",
+ database,
+ table.getTableName(),
+ dataNodeLocationMap.get(entry.getKey()));
+ setFailure(
+ new ProcedureException(
+ new MetadataException("Rollback pre-release table column extension info failed")));
return;
}
}
}
- private ByteBuffer getCacheRequestInfo() {
- ByteArrayOutputStream stream = new ByteArrayOutputStream();
- try {
- ReadWriteIOUtils.write(database, stream);
- ReadWriteIOUtils.write(tableName, stream);
-
- TsTableColumnSchemaUtil.serialize(actualAddedColumnList, stream);
- } catch (IOException ignored) {
- // won't happen
- }
- return ByteBuffer.wrap(stream.toByteArray());
- }
-
@Override
- protected AddTableColumnState getState(int stateId) {
+ protected AddTableColumnState getState(final int stateId) {
return AddTableColumnState.values()[stateId];
}
@Override
- protected int getStateId(AddTableColumnState state) {
+ protected int getStateId(final AddTableColumnState state) {
return state.ordinal();
}
@@ -287,12 +297,8 @@ public class AddTableColumnProcedure
return queryId;
}
- public List<TsTableColumnSchema> getInputColumnList() {
- return inputColumnList;
- }
-
@Override
- public void serialize(DataOutputStream stream) throws IOException {
+ public void serialize(final DataOutputStream stream) throws IOException {
stream.writeShort(ProcedureType.ADD_TABLE_COLUMN_PROCEDURE.getTypeCode());
super.serialize(stream);
@@ -300,26 +306,33 @@ public class AddTableColumnProcedure
ReadWriteIOUtils.write(tableName, stream);
ReadWriteIOUtils.write(queryId, stream);
- TsTableColumnSchemaUtil.serialize(inputColumnList, stream);
- TsTableColumnSchemaUtil.serialize(actualAddedColumnList, stream);
+ TsTableColumnSchemaUtil.serialize(addedColumnList, stream);
+ if (Objects.nonNull(table)) {
+ ReadWriteIOUtils.write(true, stream);
+ table.serialize(stream);
+ } else {
+ ReadWriteIOUtils.write(false, stream);
+ }
}
@Override
- public void deserialize(ByteBuffer byteBuffer) {
+ public void deserialize(final ByteBuffer byteBuffer) {
super.deserialize(byteBuffer);
this.database = ReadWriteIOUtils.readString(byteBuffer);
this.tableName = ReadWriteIOUtils.readString(byteBuffer);
this.queryId = ReadWriteIOUtils.readString(byteBuffer);
- this.inputColumnList = TsTableColumnSchemaUtil.deserializeColumnSchemaList(byteBuffer);
- this.actualAddedColumnList = TsTableColumnSchemaUtil.deserializeColumnSchemaList(byteBuffer);
+ this.addedColumnList = TsTableColumnSchemaUtil.deserializeColumnSchemaList(byteBuffer);
+ if (ReadWriteIOUtils.readBool(byteBuffer)) {
+ this.table = TsTable.deserialize(byteBuffer);
+ }
}
@Override
- public boolean equals(Object o) {
+ public boolean equals(final Object o) {
if (this == o) return true;
if (!(o instanceof AddTableColumnProcedure)) return false;
- AddTableColumnProcedure that = (AddTableColumnProcedure) o;
+ final AddTableColumnProcedure that = (AddTableColumnProcedure) o;
return Objects.equals(database, that.database)
&& Objects.equals(tableName, that.tableName)
&& Objects.equals(queryId, that.queryId);
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
index f3b48bb544e..f07cd4621f6 100644
--- 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
@@ -80,15 +80,15 @@ public class CreateTableProcedure
super();
}
- public CreateTableProcedure(String database, TsTable table) {
+ public CreateTableProcedure(final String database, final TsTable table) {
this.database = database;
this.table = table;
}
@Override
- protected Flow executeFromState(ConfigNodeProcedureEnv env, CreateTableState state)
+ protected Flow executeFromState(final ConfigNodeProcedureEnv env, final CreateTableState state)
throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
- long startTime = System.currentTimeMillis();
+ final long startTime = System.currentTimeMillis();
try {
switch (state) {
case PRE_CREATE:
@@ -127,12 +127,12 @@ public class CreateTableProcedure
}
}
- private void preCreateTable(ConfigNodeProcedureEnv env) {
- PreCreateTablePlan plan = new PreCreateTablePlan(database, table);
+ private void preCreateTable(final ConfigNodeProcedureEnv env) {
+ final PreCreateTablePlan plan = new PreCreateTablePlan(database, table);
TSStatus status;
try {
status = env.getConfigManager().getConsensusManager().write(plan);
- } catch (ConsensusException e) {
+ } catch (final 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());
@@ -144,19 +144,21 @@ public class CreateTableProcedure
}
}
- private void preReleaseTable(ConfigNodeProcedureEnv env) {
- TUpdateTableReq req = new TUpdateTableReq();
- req.setType(TsTableInternalRPCType.PRE_CREATE.getOperationType());
+ private void preReleaseTable(final ConfigNodeProcedureEnv env) {
+ final TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.PRE_CREATE_OR_ADD_COLUMN.getOperationType());
req.setTableInfo(TsTableInternalRPCUtil.serializeSingleTsTable(database, table));
- Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
+ final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+
if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ // All dataNodes must clear the related schema cache
LOGGER.warn(
"Failed to sync table {}.{} pre-create info to DataNode {}",
database,
@@ -169,25 +171,25 @@ public class CreateTableProcedure
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;
+ private void validateTimeSeriesExistence(final ConfigNodeProcedureEnv env) {
+ final PathPatternTree patternTree = new PathPatternTree();
+ final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ final DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ final PartialPath path;
try {
path = new PartialPath(new String[] {ROOT, database, table.getTableName()});
patternTree.appendPathPattern(path);
patternTree.appendPathPattern(path.concatNode(MULTI_LEVEL_PATH_WILDCARD));
patternTree.serialize(dataOutputStream);
- } catch (IOException e) {
+ } catch (final IOException e) {
LOGGER.warn("failed to serialize request for table {}.{}", database, table.getTableName(), e);
}
- ByteBuffer patternTreeBytes = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+ final ByteBuffer patternTreeBytes = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
- Map<TConsensusGroupId, TRegionReplicaSet> relatedSchemaRegionGroup =
+ final Map<TConsensusGroupId, TRegionReplicaSet> relatedSchemaRegionGroup =
env.getConfigManager().getRelatedSchemaRegionGroup(patternTree);
- List<TCheckTimeSeriesExistenceResp> respList = new ArrayList<>();
+ final List<TCheckTimeSeriesExistenceResp> respList = new ArrayList<>();
DataNodeRegionTaskExecutor<TCheckTimeSeriesExistenceReq, TCheckTimeSeriesExistenceResp>
regionTask =
new DataNodeRegionTaskExecutor<
@@ -201,17 +203,17 @@ public class CreateTableProcedure
@Override
protected List<TConsensusGroupId> processResponseOfOneDataNode(
- TDataNodeLocation dataNodeLocation,
- List<TConsensusGroupId> consensusGroupIdList,
- TCheckTimeSeriesExistenceResp response) {
+ final TDataNodeLocation dataNodeLocation,
+ final List<TConsensusGroupId> consensusGroupIdList,
+ final TCheckTimeSeriesExistenceResp response) {
respList.add(response);
- List<TConsensusGroupId> failedRegionList = new ArrayList<>();
+ final 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();
+ final 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()
@@ -227,7 +229,8 @@ public class CreateTableProcedure
@Override
protected void onAllReplicasetFailure(
- TConsensusGroupId consensusGroupId, Set<TDataNodeLocation> dataNodeLocationSet) {
+ final TConsensusGroupId consensusGroupId,
+ final Set<TDataNodeLocation> dataNodeLocationSet) {
setFailure(
new ProcedureException(
new MetadataException(
@@ -245,7 +248,7 @@ public class CreateTableProcedure
return;
}
- for (TCheckTimeSeriesExistenceResp resp : respList) {
+ for (final TCheckTimeSeriesExistenceResp resp : respList) {
if (resp.isExists()) {
setFailure(
new ProcedureException(
@@ -258,12 +261,12 @@ public class CreateTableProcedure
setNextState(CreateTableState.COMMIT_CREATE);
}
- private void commitCreateTable(ConfigNodeProcedureEnv env) {
- CommitCreateTablePlan plan = new CommitCreateTablePlan(database, table.getTableName());
+ private void commitCreateTable(final ConfigNodeProcedureEnv env) {
+ final CommitCreateTablePlan plan = new CommitCreateTablePlan(database, table.getTableName());
TSStatus status;
try {
status = env.getConfigManager().getConsensusManager().write(plan);
- } catch (ConsensusException e) {
+ } catch (final 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());
@@ -275,41 +278,41 @@ public class CreateTableProcedure
}
}
- private void commitReleaseTable(ConfigNodeProcedureEnv env) {
- TUpdateTableReq req = new TUpdateTableReq();
- req.setType(TsTableInternalRPCType.COMMIT_CREATE.getOperationType());
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ private void commitReleaseTable(final ConfigNodeProcedureEnv env) {
+ final TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.COMMIT_CREATE_OR_ADD_COLUMN.getOperationType());
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
try {
ReadWriteIOUtils.write(database, outputStream);
ReadWriteIOUtils.write(table.getTableName(), outputStream);
- } catch (IOException ignored) {
- //
+ } catch (final IOException ignored) {
+ // ByteArrayOutputStream will not throw IOException
}
req.setTableInfo(outputStream.toByteArray());
- Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
+ final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final 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")));
+ // TODO: Handle commit failure
return;
}
}
}
@Override
- protected void rollbackState(ConfigNodeProcedureEnv env, CreateTableState state)
+ protected void rollbackState(final ConfigNodeProcedureEnv env, final CreateTableState state)
throws IOException, InterruptedException, ProcedureException {
- long startTime = System.currentTimeMillis();
+ final long startTime = System.currentTimeMillis();
try {
switch (state) {
case PRE_CREATE:
@@ -318,7 +321,7 @@ public class CreateTableProcedure
break;
case PRE_RELEASE:
LOGGER.info("Start rollback pre release table {}.{}", database, table.getTableName());
- rollbackRelease(env);
+ rollbackPreRelease(env);
break;
}
} finally {
@@ -327,12 +330,13 @@ public class CreateTableProcedure
}
}
- private void rollbackCreate(ConfigNodeProcedureEnv env) {
- RollbackCreateTablePlan plan = new RollbackCreateTablePlan(database, table.getTableName());
+ private void rollbackCreate(final ConfigNodeProcedureEnv env) {
+ final RollbackCreateTablePlan plan =
+ new RollbackCreateTablePlan(database, table.getTableName());
TSStatus status;
try {
status = env.getConfigManager().getConsensusManager().write(plan);
- } catch (ConsensusException e) {
+ } catch (final 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());
@@ -343,25 +347,25 @@ public class CreateTableProcedure
}
}
- private void rollbackRelease(ConfigNodeProcedureEnv env) {
- TUpdateTableReq req = new TUpdateTableReq();
- req.setType(TsTableInternalRPCType.ROLLBACK_CREATE.getOperationType());
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ private void rollbackPreRelease(final ConfigNodeProcedureEnv env) {
+ final TUpdateTableReq req = new TUpdateTableReq();
+ req.setType(TsTableInternalRPCType.ROLLBACK_CREATE_OR_ADD_COLUMN.getOperationType());
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
try {
ReadWriteIOUtils.write(database, outputStream);
ReadWriteIOUtils.write(table.getTableName(), outputStream);
- } catch (IOException ignored) {
- //
+ } catch (final IOException ignore) {
+ // ByteArrayOutputStream will not throw IOException
}
req.setTableInfo(outputStream.toByteArray());
- Map<Integer, TDataNodeLocation> dataNodeLocationMap =
+ final Map<Integer, TDataNodeLocation> dataNodeLocationMap =
env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations();
- DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
+ final DataNodeAsyncRequestContext<TUpdateTableReq, TSStatus> clientHandler =
new DataNodeAsyncRequestContext<>(CnToDnRequestType.UPDATE_TABLE, req, dataNodeLocationMap);
CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
- Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
- for (Map.Entry<Integer, TSStatus> entry : statusMap.entrySet()) {
+ final Map<Integer, TSStatus> statusMap = clientHandler.getResponseMap();
+ for (final 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 {}",
@@ -375,12 +379,12 @@ public class CreateTableProcedure
}
@Override
- protected CreateTableState getState(int stateId) {
+ protected CreateTableState getState(final int stateId) {
return CreateTableState.values()[stateId];
}
@Override
- protected int getStateId(CreateTableState createTableState) {
+ protected int getStateId(final CreateTableState createTableState) {
return createTableState.ordinal();
}
@@ -398,7 +402,7 @@ public class CreateTableProcedure
}
@Override
- public void serialize(DataOutputStream stream) throws IOException {
+ public void serialize(final DataOutputStream stream) throws IOException {
stream.writeShort(ProcedureType.CREATE_TABLE_PROCEDURE.getTypeCode());
super.serialize(stream);
ReadWriteIOUtils.write(database, stream);
@@ -406,17 +410,17 @@ public class CreateTableProcedure
}
@Override
- public void deserialize(ByteBuffer byteBuffer) {
+ public void deserialize(final ByteBuffer byteBuffer) {
super.deserialize(byteBuffer);
database = ReadWriteIOUtils.readString(byteBuffer);
table = TsTable.deserialize(byteBuffer);
}
@Override
- public boolean equals(Object o) {
+ public boolean equals(final Object o) {
if (this == o) return true;
if (!(o instanceof CreateTableProcedure)) return false;
- CreateTableProcedure that = (CreateTableProcedure) o;
+ final CreateTableProcedure that = (CreateTableProcedure) o;
return Objects.equals(database, that.database) && Objects.equals(table, that.table);
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index cfa372be8dd..2a6d53c61ba 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -46,6 +46,10 @@ import org.apache.iotdb.commons.pipe.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.task.meta.PipeTaskMeta;
+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.commons.subscription.meta.consumer.ConsumerGroupMeta;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
@@ -142,6 +146,10 @@ 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.AddTableColumnPlan;
+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.CreateSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.DropSchemaTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.template.ExtendSchemaTemplatePlan;
@@ -1447,6 +1455,74 @@ public class ConfigPhysicalPlanSerDeTest {
consumerGroupHandleMetaChangePlan1.getConsumerGroupMetaList());
}
+ @Test
+ public void AddTableColumnPlanTest() throws IOException {
+ final AddTableColumnPlan addTableColumnPlan0 =
+ new AddTableColumnPlan(
+ "root.database1",
+ "table1",
+ Collections.singletonList(new IdColumnSchema("Id", TSDataType.STRING)),
+ false);
+ final AddTableColumnPlan addTableColumnPlan1 =
+ (AddTableColumnPlan)
+ ConfigPhysicalPlan.Factory.create(addTableColumnPlan0.serializeToByteBuffer());
+ Assert.assertEquals(addTableColumnPlan0.getDatabase(), addTableColumnPlan1.getDatabase());
+ Assert.assertEquals(addTableColumnPlan0.getTableName(), addTableColumnPlan1.getTableName());
+ Assert.assertEquals(
+ addTableColumnPlan0.getColumnSchemaList().size(),
+ addTableColumnPlan1.getColumnSchemaList().size());
+ Assert.assertEquals(addTableColumnPlan0.isRollback(), addTableColumnPlan1.isRollback());
+ }
+
+ @Test
+ public void CommitCreateTablePlanTest() throws IOException {
+ final CommitCreateTablePlan commitCreateTablePlan0 =
+ new CommitCreateTablePlan("root.database1", "table1");
+ final CommitCreateTablePlan commitCreateTablePlan1 =
+ (CommitCreateTablePlan)
+ ConfigPhysicalPlan.Factory.create(commitCreateTablePlan0.serializeToByteBuffer());
+ Assert.assertEquals(commitCreateTablePlan0.getDatabase(), commitCreateTablePlan1.getDatabase());
+ Assert.assertEquals(
+ commitCreateTablePlan0.getTableName(), commitCreateTablePlan1.getTableName());
+ }
+
+ @Test
+ public void preCreateTablePlanTest() throws IOException {
+ final TsTable table = new TsTable("table1");
+ table.addColumnSchema(new IdColumnSchema("Id", TSDataType.STRING));
+ table.addColumnSchema(new AttributeColumnSchema("Attr", TSDataType.STRING));
+ table.addColumnSchema(
+ new MeasurementColumnSchema(
+ "Measurement", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY));
+ final PreCreateTablePlan preCreateTablePlan0 = new PreCreateTablePlan("root.database1", table);
+ final PreCreateTablePlan preCreateTablePlan1 =
+ (PreCreateTablePlan)
+ ConfigPhysicalPlan.Factory.create(preCreateTablePlan0.serializeToByteBuffer());
+
+ Assert.assertEquals(preCreateTablePlan0.getDatabase(), preCreateTablePlan1.getDatabase());
+ Assert.assertEquals(
+ preCreateTablePlan0.getTable().getTableName(),
+ preCreateTablePlan1.getTable().getTableName());
+ Assert.assertEquals(
+ preCreateTablePlan0.getTable().getColumnNum(),
+ preCreateTablePlan1.getTable().getColumnNum());
+ Assert.assertEquals(
+ preCreateTablePlan0.getTable().getIdNums(), preCreateTablePlan1.getTable().getIdNums());
+ }
+
+ @Test
+ public void RollbackCreateTablePlanTest() throws IOException {
+ final RollbackCreateTablePlan rollbackCreateTablePlan0 =
+ new RollbackCreateTablePlan("root.database1", "table1");
+ final RollbackCreateTablePlan rollbackCreateTablePlan1 =
+ (RollbackCreateTablePlan)
+ ConfigPhysicalPlan.Factory.create(rollbackCreateTablePlan0.serializeToByteBuffer());
+ Assert.assertEquals(
+ rollbackCreateTablePlan0.getDatabase(), rollbackCreateTablePlan1.getDatabase());
+ Assert.assertEquals(
+ rollbackCreateTablePlan0.getTableName(), rollbackCreateTablePlan1.getTableName());
+ }
+
@Test
public void GetTriggerTablePlanTest() throws IOException {
GetTriggerTablePlan getTriggerTablePlan0 = new GetTriggerTablePlan(true);
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedureTest.java
new file mode 100644
index 00000000000..07a680e3cbe
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AddTableColumnProcedureTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.schema.table.column.IdColumnSchema;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+public class AddTableColumnProcedureTest {
+ @Test
+ public void serializeDeserializeTest() throws IllegalPathException, IOException {
+ final AddTableColumnProcedure addTableColumnProcedure =
+ new AddTableColumnProcedure(
+ "root.database1",
+ "table1",
+ "0",
+ Collections.singletonList(new IdColumnSchema("Id", TSDataType.STRING)));
+
+ final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ final DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ addTableColumnProcedure.serialize(dataOutputStream);
+
+ final ByteBuffer byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+
+ Assert.assertEquals(
+ ProcedureType.ADD_TABLE_COLUMN_PROCEDURE.getTypeCode(), byteBuffer.getShort());
+
+ final AddTableColumnProcedure deserializedProcedure = new AddTableColumnProcedure();
+ deserializedProcedure.deserialize(byteBuffer);
+
+ Assert.assertEquals(addTableColumnProcedure.getDatabase(), deserializedProcedure.getDatabase());
+ Assert.assertEquals(
+ addTableColumnProcedure.getTableName(), deserializedProcedure.getTableName());
+ }
+}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.java
new file mode 100644
index 00000000000..abeac0d01d5
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/CreateTableProcedureTest.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.confignode.procedure.impl.schema.table;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+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.confignode.procedure.store.ProcedureType;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class CreateTableProcedureTest {
+ @Test
+ public void serializeDeserializeTest() throws IllegalPathException, IOException {
+ final TsTable table = new TsTable("table1");
+ table.addColumnSchema(new IdColumnSchema("Id", TSDataType.STRING));
+ table.addColumnSchema(new AttributeColumnSchema("Attr", TSDataType.STRING));
+ table.addColumnSchema(
+ new MeasurementColumnSchema(
+ "Measurement", TSDataType.DOUBLE, TSEncoding.GORILLA, CompressionType.SNAPPY));
+ final CreateTableProcedure createTableProcedure =
+ new CreateTableProcedure("root.database1", table);
+
+ final ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ final DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
+ createTableProcedure.serialize(dataOutputStream);
+
+ final ByteBuffer byteBuffer = ByteBuffer.wrap(byteArrayOutputStream.toByteArray());
+
+ Assert.assertEquals(ProcedureType.CREATE_TABLE_PROCEDURE.getTypeCode(), byteBuffer.getShort());
+
+ final CreateTableProcedure deserializedProcedure = new CreateTableProcedure();
+ deserializedProcedure.deserialize(byteBuffer);
+
+ Assert.assertEquals(createTableProcedure.getDatabase(), deserializedProcedure.getDatabase());
+ Assert.assertEquals(
+ createTableProcedure.getTable().getTableName(),
+ deserializedProcedure.getTable().getTableName());
+ Assert.assertEquals(
+ createTableProcedure.getTable().getColumnNum(),
+ deserializedProcedure.getTable().getColumnNum());
+ Assert.assertEquals(
+ createTableProcedure.getTable().getIdNums(), deserializedProcedure.getTable().getIdNums());
+ }
+}
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 eeca1e43d5a..47d5c31e5d9 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
@@ -58,7 +58,6 @@ 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.table.column.TsTableColumnSchemaUtil;
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;
@@ -1443,52 +1442,31 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
}
@Override
- public TSStatus updateTable(TUpdateTableReq req) throws TException {
+ public TSStatus updateTable(TUpdateTableReq req) {
switch (TsTableInternalRPCType.getType(req.type)) {
- case PRE_CREATE:
+ case PRE_CREATE_OR_ADD_COLUMN:
DataNodeSchemaLockManager.getInstance().takeWriteLock(SchemaLockType.TIMESERIES_VS_TABLE);
try {
Pair<String, TsTable> pair =
TsTableInternalRPCUtil.deserializeSingleTsTable(req.getTableInfo());
- DataNodeTableCache.getInstance().preCreateTable(pair.left, pair.right);
+ DataNodeTableCache.getInstance().preUpdateTable(pair.left, pair.right);
} finally {
DataNodeSchemaLockManager.getInstance()
.releaseWriteLock(SchemaLockType.TIMESERIES_VS_TABLE);
}
break;
- case ROLLBACK_CREATE:
+ case ROLLBACK_CREATE_OR_ADD_COLUMN:
DataNodeTableCache.getInstance()
- .rollbackCreateTable(
+ .rollbackUpdateTable(
ReadWriteIOUtils.readString(req.tableInfo),
ReadWriteIOUtils.readString(req.tableInfo));
break;
- case COMMIT_CREATE:
+ case COMMIT_CREATE_OR_ADD_COLUMN:
DataNodeTableCache.getInstance()
- .commitCreateTable(
+ .commitUpdateTable(
ReadWriteIOUtils.readString(req.tableInfo),
ReadWriteIOUtils.readString(req.tableInfo));
break;
- case PRE_ADD_COLUMN:
- DataNodeTableCache.getInstance()
- .preAddTableColumn(
- ReadWriteIOUtils.readString(req.tableInfo),
- ReadWriteIOUtils.readString(req.tableInfo),
- TsTableColumnSchemaUtil.deserializeColumnSchemaList(req.tableInfo));
- break;
- case COMMIT_ADD_COLUMN:
- DataNodeTableCache.getInstance()
- .commitAddTableColumn(
- ReadWriteIOUtils.readString(req.tableInfo),
- ReadWriteIOUtils.readString(req.tableInfo),
- TsTableColumnSchemaUtil.deserializeColumnSchemaList(req.tableInfo));
- break;
- case ROLLBACK_ADD_COLUMN:
- DataNodeTableCache.getInstance()
- .rollbackAddColumn(
- ReadWriteIOUtils.readString(req.tableInfo),
- ReadWriteIOUtils.readString(req.tableInfo),
- TsTableColumnSchemaUtil.deserializeColumnSchemaList(req.tableInfo));
- break;
default:
LOGGER.warn("Unsupported type {} when updating table", req.type);
return RpcUtils.getStatus(TSStatusCode.ILLEGAL_PARAMETER);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
index 3aa287c8f5b..60413cc579f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java
@@ -22,7 +22,6 @@ 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.commons.schema.table.column.TsTableColumnSchema;
import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
@@ -45,15 +44,12 @@ public class DataNodeTableCache implements ITableCache {
private final Map<String, Map<String, TsTable>> databaseTableMap = new ConcurrentHashMap<>();
- private final Map<String, Map<String, TsTable>> preCreateTableMap = new ConcurrentHashMap<>();
-
- private final Map<String, Map<String, List<TsTableColumnSchema>>> preAddColumnMap =
- new ConcurrentHashMap<>();
+ private final Map<String, Map<String, TsTable>> preUpdateTableMap = new ConcurrentHashMap<>();
private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private DataNodeTableCache() {
- // do nothing
+ // Do nothing
}
private static final class DataNodeTableCacheHolder {
@@ -67,18 +63,18 @@ public class DataNodeTableCache implements ITableCache {
}
@Override
- public void init(byte[] tableInitializationBytes) {
+ public void init(final byte[] tableInitializationBytes) {
readWriteLock.writeLock().lock();
try {
if (tableInitializationBytes == null) {
return;
}
- Pair<Map<String, List<TsTable>>, Map<String, List<TsTable>>> tableInfo =
+ final 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;
+ final Map<String, List<TsTable>> usingMap = tableInfo.left;
+ final Map<String, List<TsTable>> preCreateMap = tableInfo.right;
saveUpdatedTableInfo(usingMap, databaseTableMap);
- saveUpdatedTableInfo(preCreateMap, preCreateTableMap);
+ saveUpdatedTableInfo(preCreateMap, preUpdateTableMap);
LOGGER.info("Init DataNodeTableCache successfully");
} finally {
readWriteLock.writeLock().unlock();
@@ -86,10 +82,11 @@ public class DataNodeTableCache implements ITableCache {
}
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()) {
+ final Map<String, List<TsTable>> tableMap,
+ final Map<String, Map<String, TsTable>> localTableMap) {
+ for (final Map.Entry<String, List<TsTable>> entry : tableMap.entrySet()) {
+ final Map<String, TsTable> map = new ConcurrentHashMap<>();
+ for (final TsTable table : entry.getValue()) {
map.put(table.getTableName(), table);
}
localTableMap.put(entry.getKey(), map);
@@ -97,31 +94,31 @@ public class DataNodeTableCache implements ITableCache {
}
@Override
- public void preCreateTable(String database, TsTable table) {
+ public void preUpdateTable(final String database, final TsTable table) {
readWriteLock.writeLock().lock();
try {
- preCreateTableMap
+ preUpdateTableMap
.computeIfAbsent(database, k -> new ConcurrentHashMap<>())
.put(table.getTableName(), table);
- LOGGER.info("Pre-create table {}.{} successfully", database, table);
+ LOGGER.info("Pre-update table {}.{} successfully", database, table);
} finally {
readWriteLock.writeLock().unlock();
}
}
@Override
- public void rollbackCreateTable(String database, String tableName) {
+ public void rollbackUpdateTable(final String database, final String tableName) {
readWriteLock.writeLock().lock();
try {
- removeTableFromPreCreateMap(database, tableName);
- LOGGER.info("Rollback-create table {}.{} successfully", database, tableName);
+ removeTableFromPreUpdateMap(database, tableName);
+ LOGGER.info("Rollback-update table {}.{} successfully", database, tableName);
} finally {
readWriteLock.writeLock().unlock();
}
}
- private void removeTableFromPreCreateMap(String database, String tableName) {
- preCreateTableMap.compute(
+ private void removeTableFromPreUpdateMap(final String database, final String tableName) {
+ preUpdateTableMap.compute(
database,
(k, v) -> {
if (v == null) {
@@ -137,93 +134,32 @@ public class DataNodeTableCache implements ITableCache {
}
@Override
- public void commitCreateTable(String database, String tableName) {
+ public void commitUpdateTable(final String database, final String tableName) {
readWriteLock.writeLock().lock();
try {
- TsTable table = preCreateTableMap.get(database).get(tableName);
+ final TsTable table = preUpdateTableMap.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();
- }
- }
-
- @Override
- public void preAddTableColumn(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList) {
- readWriteLock.writeLock().lock();
- try {
- preAddColumnMap
- .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
- .computeIfAbsent(tableName, k -> new ArrayList<>())
- .addAll(columnSchemaList);
- } finally {
- readWriteLock.writeLock().unlock();
- }
- }
-
- @Override
- public void commitAddTableColumn(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList) {
- readWriteLock.writeLock().lock();
- try {
- TsTable table = databaseTableMap.get(database).get(tableName);
- columnSchemaList.forEach(table::addColumnSchema);
- preAddColumnMap.compute(
- database,
- (k, v) -> {
- if (v == null) {
- throw new IllegalStateException();
- }
- v.remove(tableName);
- if (v.isEmpty()) {
- return null;
- }
- return v;
- });
- } finally {
- readWriteLock.writeLock().unlock();
- }
- }
-
- @Override
- public void rollbackAddColumn(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList) {
- readWriteLock.writeLock().lock();
- try {
- preAddColumnMap.compute(
- database,
- (k, v) -> {
- if (v == null) {
- throw new IllegalStateException();
- }
- v.remove(tableName);
- if (v.isEmpty()) {
- return null;
- }
- return v;
- });
+ removeTableFromPreUpdateMap(database, tableName);
+ LOGGER.info("Commit-update table {}.{} successfully", database, tableName);
} finally {
readWriteLock.writeLock().unlock();
}
}
@Override
- public void invalid(String database) {
+ public void invalid(final String database) {
readWriteLock.writeLock().lock();
try {
databaseTableMap.remove(database);
- preCreateTableMap.remove(database);
- preAddColumnMap.remove(database);
+ preUpdateTableMap.remove(database);
} finally {
readWriteLock.writeLock().unlock();
}
}
- public TsTable getTable(String database, String tableName) {
+ public TsTable getTable(final String database, final String tableName) {
readWriteLock.readLock().lock();
try {
if (databaseTableMap.containsKey(database)) {
@@ -235,10 +171,10 @@ public class DataNodeTableCache implements ITableCache {
}
}
- public Optional<List<TsTable>> getTables(String database) {
+ public Optional<List<TsTable>> getTables(final String database) {
readWriteLock.readLock().lock();
try {
- Map<String, TsTable> tableMap = databaseTableMap.get(database);
+ final Map<String, TsTable> tableMap = databaseTableMap.get(database);
return tableMap != null ? Optional.of(new ArrayList<>(tableMap.values())) : Optional.empty();
} finally {
readWriteLock.readLock().unlock();
@@ -246,13 +182,13 @@ public class DataNodeTableCache implements ITableCache {
}
/** Check whether the given path overlap with some table existence. */
- public Pair<String, String> checkTableCreateAndPreCreateOnGivenPath(PartialPath path) {
+ public Pair<String, String> checkTableCreateAndPreCreateOnGivenPath(final PartialPath path) {
readWriteLock.writeLock().lock();
try {
- String pathString = path.getFullPath();
+ final String pathString = path.getFullPath();
Pair<String, String> result = checkTableExistenceOnGivenPath(pathString, databaseTableMap);
if (result == null) {
- result = checkTableExistenceOnGivenPath(pathString, preCreateTableMap);
+ result = checkTableExistenceOnGivenPath(pathString, preUpdateTableMap);
}
return result;
} finally {
@@ -261,16 +197,16 @@ public class DataNodeTableCache implements ITableCache {
}
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();
+ final String path, final Map<String, Map<String, TsTable>> tableMap) {
+ final int dbStartIndex = PATH_ROOT.length() + 1;
+ for (final Map.Entry<String, Map<String, TsTable>> dbEntry : tableMap.entrySet()) {
+ final 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()) {
+ final int tableStartIndex = dbStartIndex + database.length() + 1;
+ for (final String tableName : dbEntry.getValue().keySet()) {
if (path.startsWith(tableName, tableStartIndex)
&& path.charAt(tableStartIndex + tableName.length()) == PATH_SEPARATOR) {
return new Pair<>(database, tableName);
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
index 79d28783efc..fbc50d6f4de 100644
--- 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
@@ -20,31 +20,19 @@
package org.apache.iotdb.db.schemaengine.table;
import org.apache.iotdb.commons.schema.table.TsTable;
-import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
-
-import java.util.List;
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);
+ void init(final byte[] tableInitializationBytes);
- void preAddTableColumn(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList);
+ void preUpdateTable(final String database, final TsTable table);
- void commitAddTableColumn(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList);
+ void rollbackUpdateTable(final String database, final String tableName);
- void rollbackAddColumn(
- String database, String tableName, List<TsTableColumnSchema> columnSchemaList);
+ void commitUpdateTable(final String database, final String tableName);
/**
* @param database shouldn't start with `root.`
*/
- void invalid(String database);
+ void invalid(final String database);
}
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
index 2ff0edcc633..e76160e7f99 100644
--- 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
@@ -26,17 +26,13 @@ import java.io.OutputStream;
import java.nio.ByteBuffer;
public enum TsTableInternalRPCType {
- PRE_CREATE((byte) 0),
- ROLLBACK_CREATE((byte) 1),
- COMMIT_CREATE((byte) 2),
-
- PRE_ADD_COLUMN((byte) 6),
- COMMIT_ADD_COLUMN((byte) 7),
- ROLLBACK_ADD_COLUMN((byte) 8);
+ PRE_CREATE_OR_ADD_COLUMN((byte) 0),
+ ROLLBACK_CREATE_OR_ADD_COLUMN((byte) 1),
+ COMMIT_CREATE_OR_ADD_COLUMN((byte) 2);
private final byte operationType;
- private TsTableInternalRPCType(byte operationType) {
+ private TsTableInternalRPCType(final byte operationType) {
this.operationType = operationType;
}
@@ -44,29 +40,23 @@ public enum TsTableInternalRPCType {
return operationType;
}
- public void serialize(OutputStream stream) throws IOException {
+ public void serialize(final OutputStream stream) throws IOException {
ReadWriteIOUtils.write(operationType, stream);
}
- public static TsTableInternalRPCType deserialize(ByteBuffer buffer) {
+ public static TsTableInternalRPCType deserialize(final ByteBuffer buffer) {
byte type = ReadWriteIOUtils.readByte(buffer);
return getType(type);
}
- public static TsTableInternalRPCType getType(byte type) {
+ public static TsTableInternalRPCType getType(final byte type) {
switch (type) {
case 0:
- return PRE_CREATE;
+ return PRE_CREATE_OR_ADD_COLUMN;
case 1:
- return ROLLBACK_CREATE;
+ return ROLLBACK_CREATE_OR_ADD_COLUMN;
case 2:
- return COMMIT_CREATE;
- case 6:
- return PRE_ADD_COLUMN;
- case 7:
- return COMMIT_ADD_COLUMN;
- case 8:
- return ROLLBACK_ADD_COLUMN;
+ return COMMIT_CREATE_OR_ADD_COLUMN;
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
index d98fc3e01a2..ac9fa72d8e4 100644
--- 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
@@ -77,25 +77,25 @@ public class TsTableInternalRPCUtil {
return result;
}
- public static byte[] serializeSingleTsTable(String database, TsTable table) {
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ public static byte[] serializeSingleTsTable(final String database, final TsTable table) {
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
try {
ReadWriteIOUtils.write(database, outputStream);
table.serialize(outputStream);
- } catch (IOException ignored) {
- // won't happen
+ } catch (final IOException ignored) {
+ // ByteArrayOutputStream won't throw IOException
}
return outputStream.toByteArray();
}
- public static Pair<String, TsTable> deserializeSingleTsTable(byte[] bytes) {
+ public static Pair<String, TsTable> deserializeSingleTsTable(final 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
+ // ByteArrayInputStream won't throw IOException
}
throw new IllegalStateException();
}
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
index 30b006a3fbd..c227f96d705 100644
--- 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
@@ -32,7 +32,7 @@ import java.util.List;
public class TsTableColumnSchemaUtil {
private TsTableColumnSchemaUtil() {
- // do nothing
+ // Do nothing
}
public static void serialize(TsTableColumnSchema columnSchema, OutputStream outputStream)
@@ -86,7 +86,7 @@ public class TsTableColumnSchemaUtil {
try {
serialize(columnSchemaList, stream);
} catch (IOException ignored) {
- // won't happen
+ // ByteArrayOutputStream won't throw IOException
}
return stream.toByteArray();
}