This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 3b47945b3ce Show pre-deleted table schemas and improve write errors
(#18626)
3b47945b3ce is described below
commit 3b47945b3ceb0c3c25c1951a7afd3b2424fde3d5
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 21 11:52:09 2026 +0800
Show pre-deleted table schemas and improve write errors (#18626)
* Show pre-deleted table schemas and improve write errors
* Support rollback for pre-altered table columns
* Fix TreeViewTest metadata lease timeout
* Clarify pre-deleted column cache scope
---
.../consensus/request/ConfigPhysicalPlan.java | 4 +
.../consensus/request/ConfigPhysicalPlanType.java | 1 +
.../table/RollbackPreAlterColumnDataTypePlan.java | 61 ++++
.../manager/schema/ClusterSchemaManager.java | 40 ++-
.../persistence/executor/ConfigPlanExecutor.java | 4 +
.../persistence/schema/ClusterSchemaInfo.java | 103 +++++-
.../confignode/persistence/schema/ConfigMTree.java | 169 +++++++++-
.../table/AlterTableColumnDataTypeProcedure.java | 78 ++++-
.../impl/schema/table/CreateTableProcedure.java | 16 +-
.../request/ConfigPhysicalPlanSerDeTest.java | 15 +
.../persistence/schema/ConfigMTreeTest.java | 77 ++++-
.../persistence/schema/TablePreDeleteTest.java | 368 +++++++++++++++++++++
.../AlterTableColumnDataTypeProcedureTest.java | 171 ++++++++++
.../schema/table/CreateTableProcedureTest.java | 39 +++
.../iotdb/db/i18n/DataNodeSchemaMessages.java | 2 -
.../iotdb/db/i18n/DataNodeSchemaMessages.java | 2 -
.../config/executor/ClusterConfigTaskExecutor.java | 14 +
.../fetcher/TableHeaderSchemaValidator.java | 41 ++-
.../db/schemaengine/table/DataNodeTableCache.java | 86 ++++-
.../plan/relational/analyzer/TreeViewTest.java | 5 +
.../fetcher/TableHeaderSchemaValidatorTest.java | 251 ++++++++++++++
.../schemaengine/table/DataNodeTableCacheTest.java | 191 ++++++++++-
.../apache/iotdb/commons/i18n/CommonMessages.java | 6 +
.../apache/iotdb/commons/i18n/CommonMessages.java | 6 +
.../exception/table/ColumnInAlterException.java | 40 +++
.../exception/table/ColumnInDeletionException.java | 40 +++
.../exception/table/TableInDeletionException.java | 38 +++
27 files changed, 1819 insertions(+), 49 deletions(-)
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 0d9ca912571..ff765a035b0 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
@@ -116,6 +116,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTableP
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreDeleteTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan;
@@ -463,6 +464,9 @@ public abstract class ConfigPhysicalPlan implements
IConsensusRequest {
case PreAlterColumnDataType:
plan = new PreAlterColumnDataTypePlan();
break;
+ case RollbackPreAlterColumnDataType:
+ plan = new RollbackPreAlterColumnDataTypePlan();
+ break;
case AlterColumnDataType:
plan = new AlterColumnDataTypePlan();
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 1be95181414..afef9f82bfe 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
@@ -233,6 +233,7 @@ public enum ConfigPhysicalPlanType {
AlterColumnDataType((short) 878),
PreAlterColumnDataType((short) 879),
RollbackPreDeleteTable((short) 880),
+ RollbackPreAlterColumnDataType((short) 881),
/** Deprecated types for sync, restored them for upgrade. */
@Deprecated
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java
new file mode 100644
index 00000000000..bcb9fb4ca72
--- /dev/null
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/table/RollbackPreAlterColumnDataTypePlan.java
@@ -0,0 +1,61 @@
+/*
+ * 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.ConfigPhysicalPlanType;
+
+import org.apache.tsfile.enums.TSDataType;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+public class RollbackPreAlterColumnDataTypePlan extends
AbstractTableColumnPlan {
+ private TSDataType newType;
+
+ public RollbackPreAlterColumnDataTypePlan() {
+ super(ConfigPhysicalPlanType.RollbackPreAlterColumnDataType);
+ }
+
+ public RollbackPreAlterColumnDataTypePlan(
+ final String database,
+ final String tableName,
+ final String columnName,
+ final TSDataType newType) {
+ super(ConfigPhysicalPlanType.RollbackPreAlterColumnDataType, database,
tableName, columnName);
+ this.newType = newType;
+ }
+
+ @Override
+ protected void serializeImpl(final DataOutputStream stream) throws
IOException {
+ super.serializeImpl(stream);
+ stream.write(newType.serialize());
+ }
+
+ @Override
+ protected void deserializeImpl(final ByteBuffer buffer) throws IOException {
+ super.deserializeImpl(buffer);
+ newType = TSDataType.deserializeFrom(buffer);
+ }
+
+ public TSDataType getNewType() {
+ return newType;
+ }
+}
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 840be42d48c..b55180e35f7 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
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.schema.SchemaConstant;
@@ -1540,13 +1541,34 @@ public class ClusterSchemaManager {
return clusterSchemaInfo.getTsTableIfExists(database, tableName);
}
+ public boolean isColumnAlterCommitted(
+ final String database,
+ final String tableName,
+ final String columnName,
+ final TSDataType dataType)
+ throws MetadataException {
+ return clusterSchemaInfo.isColumnAlterCommitted(database, tableName,
columnName, dataType);
+ }
+
+ public Optional<TSDataType> getPreAlteredColumnType(
+ final String database, final String tableName, final String columnName)
+ throws MetadataException {
+ return clusterSchemaInfo.getPreAlteredColumnType(database, tableName,
columnName);
+ }
+
public synchronized Pair<TSStatus, TsTable>
tableColumnCheckForColumnExtension(
final String database,
final String tableName,
final List<TsTableColumnSchema> columnSchemaList,
final boolean isTableView)
throws MetadataException {
- final TsTable originalTable = getTableIfExists(database,
tableName).orElse(null);
+ final TsTable originalTable =
+ clusterSchemaInfo.getTableForModification(
+ database,
+ tableName,
+ columnSchemaList.stream()
+ .map(TsTableColumnSchema::getColumnName)
+ .toArray(String[]::new));
if (Objects.isNull(originalTable)) {
return new Pair<>(
@@ -1601,7 +1623,7 @@ public class ClusterSchemaManager {
final TSDataType dataType,
final boolean isGeneratedByPipe)
throws MetadataException {
- final TsTable originalTable = getTableIfExists(database,
tableName).orElse(null);
+ final TsTable originalTable =
clusterSchemaInfo.getTableForModification(database, tableName);
if (Objects.isNull(originalTable)) {
return new Pair<>(
@@ -1638,7 +1660,8 @@ public class ClusterSchemaManager {
final String newName,
final boolean isTableView)
throws MetadataException {
- final TsTable originalTable = getTableIfExists(database,
tableName).orElse(null);
+ final TsTable originalTable =
+ clusterSchemaInfo.getTableForModification(database, tableName,
oldName, newName);
if (Objects.isNull(originalTable)) {
return new Pair<>(
@@ -1691,7 +1714,7 @@ public class ClusterSchemaManager {
final String newName,
final boolean isTableView)
throws MetadataException {
- final TsTable originalTable = getTableIfExists(database,
tableName).orElse(null);
+ final TsTable originalTable =
clusterSchemaInfo.getTableForModification(database, tableName);
if (Objects.isNull(originalTable)) {
return new Pair<>(
@@ -1707,7 +1730,12 @@ public class ClusterSchemaManager {
return result.get();
}
- if (getTableIfExists(database, newName).isPresent()) {
+ final Optional<Pair<TsTable, TableNodeStatus>> targetTable =
+ getTableAndStatusIfExists(database, newName);
+ if (targetTable.isPresent() && targetTable.get().getRight() ==
TableNodeStatus.PRE_DELETE) {
+ throw new TableInDeletionException(database, newName);
+ }
+ if (targetTable.isPresent()) {
return new Pair<>(
RpcUtils.getStatus(
TSStatusCode.TABLE_ALREADY_EXISTS,
@@ -1776,7 +1804,7 @@ public class ClusterSchemaManager {
final Map<String, String> updatedProperties,
final boolean isTableView)
throws MetadataException {
- final TsTable originalTable = getTableIfExists(database,
tableName).orElse(null);
+ final TsTable originalTable =
clusterSchemaInfo.getTableForModification(database, tableName);
if (Objects.isNull(originalTable)) {
return new Pair<>(
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 772f46baa31..f2d6ae80926 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
@@ -134,6 +134,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTableP
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreDeleteTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan;
@@ -615,6 +616,9 @@ public class ConfigPlanExecutor {
return clusterSchemaInfo.dropTable((CommitDeleteTablePlan)
physicalPlan);
case PreAlterColumnDataType:
return
clusterSchemaInfo.preAlterColumnDataType((PreAlterColumnDataTypePlan)
physicalPlan);
+ case RollbackPreAlterColumnDataType:
+ return clusterSchemaInfo.rollbackPreAlterColumnDataType(
+ (RollbackPreAlterColumnDataTypePlan) physicalPlan);
case AlterColumnDataType:
return clusterSchemaInfo.commitAlterColumnDataType(
((AlterColumnDataTypePlan) physicalPlan));
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 1b9c880cd6a..dc8106a76a6 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
@@ -27,6 +27,9 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.exception.table.ColumnInAlterException;
+import org.apache.iotdb.commons.exception.table.ColumnInDeletionException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
import org.apache.iotdb.commons.schema.table.TableNodeStatus;
@@ -34,6 +37,7 @@ import org.apache.iotdb.commons.schema.table.TableType;
import org.apache.iotdb.commons.schema.table.TreeViewSchema;
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.iotdb.commons.schema.template.Template;
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
import org.apache.iotdb.commons.utils.PathUtils;
@@ -67,6 +71,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTableP
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreDeleteTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan;
@@ -109,6 +114,7 @@ import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.annotations.TableModel;
+import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1352,9 +1358,11 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
})
.collect(Collectors.toList())
: tableModelMTree
- .getAllUsingTablesUnderSpecificDatabase(
+ .getAllTablesUnderSpecificDatabase(
getQualifiedDatabasePartialPath(plan.getDatabase()))
.stream()
+ .filter(pair -> pair.getRight() !=
TableNodeStatus.PRE_CREATE)
+ .map(Pair::getLeft)
.map(
tsTable ->
new TTableInfo(
@@ -1447,7 +1455,7 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
}
return new DescTableResp(
StatusUtils.OK,
- tableModelMTree.getUsingTableSchema(databasePath,
plan.getTableName()),
+ tableModelMTree.getTableSchemaForDesc(databasePath,
plan.getTableName()),
null,
null);
} catch (final MetadataException e) {
@@ -1557,6 +1565,80 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
}
}
+ public TsTable getTableForModification(
+ final String database, final String tableName, final String...
columnNames)
+ throws MetadataException {
+ databaseReadWriteLock.readLock().lock();
+ try {
+ final PartialPath databasePath =
getQualifiedDatabasePartialPath(database);
+ final Optional<Pair<TsTable, TableNodeStatus>> tableAndStatus =
+ tableModelMTree.getTableAndStatusIfExists(databasePath, tableName);
+ if (!tableAndStatus.isPresent()) {
+ return null;
+ }
+ if (tableAndStatus.get().getRight() == TableNodeStatus.PRE_DELETE) {
+ throw new TableInDeletionException(database, tableName);
+ }
+ final TableSchemaDetails details =
+ tableModelMTree.getTableSchemaDetails(databasePath, tableName);
+ for (final String columnName : columnNames) {
+ if (details.preDeletedColumns.contains(columnName)) {
+ throw new ColumnInDeletionException(database, tableName, columnName);
+ }
+ if (details.preAlteredColumns.containsKey(columnName)) {
+ throw new ColumnInAlterException(database, tableName, columnName);
+ }
+ }
+ return tableModelMTree.getTableSchemaForDataNode(databasePath,
tableName);
+ } finally {
+ databaseReadWriteLock.readLock().unlock();
+ }
+ }
+
+ public boolean isColumnAlterCommitted(
+ final String database,
+ final String tableName,
+ final String columnName,
+ final TSDataType dataType)
+ throws MetadataException {
+ databaseReadWriteLock.readLock().lock();
+ try {
+ final PartialPath databasePath =
getQualifiedDatabasePartialPath(database);
+ final Optional<Pair<TsTable, TableNodeStatus>> tableAndStatus =
+ tableModelMTree.getTableAndStatusIfExists(databasePath, tableName);
+ if (!tableAndStatus.isPresent()) {
+ return false;
+ }
+ final TableSchemaDetails details =
+ tableModelMTree.getTableSchemaDetails(databasePath, tableName);
+ final TsTableColumnSchema columnSchema =
details.table.getColumnSchema(columnName);
+ return !details.preAlteredColumns.containsKey(columnName)
+ && columnSchema != null
+ && columnSchema.getDataType() == dataType;
+ } finally {
+ databaseReadWriteLock.readLock().unlock();
+ }
+ }
+
+ public Optional<TSDataType> getPreAlteredColumnType(
+ final String database, final String tableName, final String columnName)
+ throws MetadataException {
+ databaseReadWriteLock.readLock().lock();
+ try {
+ final PartialPath databasePath =
getQualifiedDatabasePartialPath(database);
+ if (!tableModelMTree.getTableAndStatusIfExists(databasePath,
tableName).isPresent()) {
+ return Optional.empty();
+ }
+ return Optional.ofNullable(
+ tableModelMTree
+ .getTableSchemaDetails(databasePath, tableName)
+ .preAlteredColumns
+ .get(columnName));
+ } finally {
+ databaseReadWriteLock.readLock().unlock();
+ }
+ }
+
public TSStatus addTableColumn(final AddTableColumnPlan plan) {
return executeWithLock(
() -> {
@@ -1644,6 +1726,23 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
}
}
+ public TSStatus rollbackPreAlterColumnDataType(final
RollbackPreAlterColumnDataTypePlan plan) {
+ databaseReadWriteLock.writeLock().lock();
+ try {
+ tableModelMTree.rollbackPreAlterColumnDataType(
+ getQualifiedDatabasePartialPath(plan.getDatabase()),
+ plan.getTableName(),
+ plan.getColumnName(),
+ plan.getNewType());
+ return RpcUtils.SUCCESS_STATUS;
+ } catch (final MetadataException e) {
+ LOGGER.warn(e.getMessage(), e);
+ return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
+ } finally {
+ databaseReadWriteLock.writeLock().unlock();
+ }
+ }
+
public TSStatus commitAlterColumnDataType(AlterColumnDataTypePlan plan) {
databaseReadWriteLock.writeLock().lock();
try {
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 12f6a2baaa1..53dacbd7196 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
@@ -25,8 +25,11 @@ 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.exception.SemanticException;
+import org.apache.iotdb.commons.exception.table.ColumnInAlterException;
+import org.apache.iotdb.commons.exception.table.ColumnInDeletionException;
import org.apache.iotdb.commons.exception.table.ColumnNotExistsException;
import org.apache.iotdb.commons.exception.table.TableAlreadyExistsException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
import org.apache.iotdb.commons.exception.table.TableNotExistsException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PathPatternTree;
@@ -684,6 +687,9 @@ public class ConfigMTree {
tableNode.setTable(table);
tableNode.setStatus(TableNodeStatus.PRE_CREATE);
} else if (node instanceof ConfigTableNode) {
+ if (((ConfigTableNode) node).getStatus() == TableNodeStatus.PRE_DELETE) {
+ throw new TableInDeletionException(database.getFullPath(),
table.getTableName());
+ }
throw new TableAlreadyExistsException(
database.getFullPath().substring(ROOT.length() + 1),
table.getTableName());
} else {
@@ -697,6 +703,10 @@ public class ConfigMTree {
final IConfigMNode databaseNode =
getDatabaseNodeByDatabasePath(database).getAsMNode();
final IConfigMNode node = databaseNode.getChild(table.getTableName());
if (Objects.nonNull(node)) {
+ if (node instanceof ConfigTableNode
+ && ((ConfigTableNode) node).getStatus() ==
TableNodeStatus.PRE_DELETE) {
+ throw new TableInDeletionException(database.getFullPath(),
table.getTableName());
+ }
if (!TreeViewSchema.isTreeViewTable(((ConfigTableNode)
node).getTable())) {
throw new TableAlreadyExistsException(
database.getFullPath().substring(ROOT.length() + 1),
table.getTableName());
@@ -774,7 +784,7 @@ public class ConfigMTree {
public void renameTable(final PartialPath database, final String tableName,
final String newName)
throws MetadataException {
final IConfigMNode databaseNode =
getDatabaseNodeByDatabasePath(database).getAsMNode();
- final ConfigTableNode tableNode = (ConfigTableNode)
databaseNode.getChild(tableName);
+ final ConfigTableNode tableNode = getTableNodeForModification(database,
tableName);
store.deleteChild(databaseNode, tableName);
tableNode.setName(newName);
store.addChild(databaseNode, newName, tableNode);
@@ -786,7 +796,19 @@ public class ConfigMTree {
final String oldName,
final String newName)
throws MetadataException {
- final ConfigTableNode tableNode = getTableNode(database, tableName);
+ final ConfigTableNode tableNode = getTableNodeForModification(database,
tableName);
+ if (tableNode.getPreDeletedColumns().contains(oldName)) {
+ throw new ColumnInDeletionException(database.getFullPath(), tableName,
oldName);
+ }
+ if (tableNode.getPreDeletedColumns().contains(newName)) {
+ throw new ColumnInDeletionException(database.getFullPath(), tableName,
newName);
+ }
+ if (tableNode.getPreAlteredColumns().containsKey(oldName)) {
+ throw new ColumnInAlterException(database.getFullPath(), tableName,
oldName);
+ }
+ if (tableNode.getPreAlteredColumns().containsKey(newName)) {
+ throw new ColumnInAlterException(database.getFullPath(), tableName,
newName);
+ }
tableNode.getTable().renameColumnSchema(oldName, newName);
}
@@ -796,7 +818,7 @@ public class ConfigMTree {
final String comment,
final boolean isView)
throws MetadataException {
- final TsTable table = getTable(database, tableName);
+ final TsTable table = getTableForModification(database, tableName);
final Optional<Pair<TSStatus, TsTable>> check =
ClusterSchemaManager.checkTable4View(database.getTailNode(), table,
isView);
if (check.isPresent()) {
@@ -816,7 +838,8 @@ public class ConfigMTree {
final @Nonnull String columnName,
final @Nullable String comment)
throws MetadataException {
- final TsTable table = getTable(database, tableName);
+ final ConfigTableNode node = getTableNodeForModification(database,
tableName);
+ final TsTable table = node.getTable();
final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName);
@@ -824,6 +847,12 @@ public class ConfigMTree {
throw new ColumnNotExistsException(
PathUtils.unQualifyDatabaseName(database.getFullPath()), tableName,
columnName);
}
+ if (node.getPreDeletedColumns().contains(columnName)) {
+ throw new ColumnInDeletionException(database.getFullPath(), tableName,
columnName);
+ }
+ if (node.getPreAlteredColumns().containsKey(columnName)) {
+ throw new ColumnInAlterException(database.getFullPath(), tableName,
columnName);
+ }
if (Objects.nonNull(comment)) {
columnSchema.getProps().put(TsTable.COMMENT_KEY, comment);
} else {
@@ -839,7 +868,7 @@ public class ConfigMTree {
child ->
child instanceof ConfigTableNode
&& ((ConfigTableNode)
child).getStatus().equals(TableNodeStatus.USING))
- .map(child -> ((ConfigTableNode) child).getTable())
+ .map(child -> getTableSchemaForDataNode((ConfigTableNode) child))
.collect(Collectors.toList());
}
@@ -869,7 +898,7 @@ public class ConfigMTree {
TsTable table =
((ConfigTableNode) child).getStatus() == TableNodeStatus.PRE_DELETE
? new PreDeleteTsTable(tableName)
- : ((ConfigTableNode) child).getTable();
+ : getTableSchemaForDataNode((ConfigTableNode) child);
result.put(tableName, table);
} else {
result.put(tableName, null);
@@ -939,7 +968,13 @@ public class ConfigMTree {
final String tableName,
final List<TsTableColumnSchema> columnSchemaList)
throws MetadataException {
- final TsTable table = getTable(database, tableName);
+ final TsTable table =
+ getTableForModification(
+ database,
+ tableName,
+ columnSchemaList.stream()
+ .map(TsTableColumnSchema::getColumnName)
+ .toArray(String[]::new));
columnSchemaList.forEach(table::addColumnSchema);
}
@@ -960,7 +995,8 @@ public class ConfigMTree {
throw new TableNotExistsException(
database.getFullPath().substring(ROOT.length() + 1), tableName);
}
- final TsTable table = ((ConfigTableNode)
databaseNode.getChild(tableName)).getTable();
+ final ConfigTableNode tableNode = getTableNodeForModification(database,
tableName);
+ final TsTable table = tableNode.getTable();
tableProperties.forEach(
(k, v) -> {
if (Objects.nonNull(v)) {
@@ -988,7 +1024,7 @@ public class ConfigMTree {
final String columnName,
final boolean isView)
throws MetadataException, SemanticException {
- final ConfigTableNode node = getTableNode(database, tableName);
+ final ConfigTableNode node = getTableNodeForModification(database,
tableName);
final Optional<Pair<TSStatus, TsTable>> check =
ClusterSchemaManager.checkTable4View(database.getTailNode(),
node.getTable(), isView);
if (check.isPresent()) {
@@ -1006,6 +1042,10 @@ public class ConfigMTree {
throw new
SemanticException(ConfigNodeMessages.DROPPING_TAG_OR_TIME_COLUMN_IS_NOT_SUPPORTED);
}
+ if (node.getPreAlteredColumns().containsKey(columnName)) {
+ throw new ColumnInAlterException(database.getFullPath(), tableName,
columnName);
+ }
+
node.addPreDeletedColumn(columnName);
return columnSchema.getColumnCategory() == TsTableColumnCategory.ATTRIBUTE;
}
@@ -1018,22 +1058,34 @@ public class ConfigMTree {
if (Objects.nonNull(table.getColumnSchema(columnName))) {
table.removeColumnSchema(columnName);
node.removePreDeletedColumn(columnName);
+ node.removePreAlteredColumn(columnName);
}
}
public void preAlterColumnDataType(
PartialPath database, String tableName, String columnName, TSDataType
dataType)
throws MetadataException {
- final ConfigTableNode node = getTableNode(database, tableName);
+ final ConfigTableNode node = getTableNodeForModification(database,
tableName);
final TsTableColumnSchema columnSchema =
node.getTable().getColumnSchema(columnName);
if (Objects.isNull(columnSchema)) {
throw new ColumnNotExistsException(
PathUtils.unQualifyDatabaseName(database.getFullPath()), tableName,
columnName);
}
+ if (node.getPreDeletedColumns().contains(columnName)) {
+ throw new ColumnInDeletionException(database.getFullPath(), tableName,
columnName);
+ }
if (columnSchema.getColumnCategory() != TsTableColumnCategory.FIELD) {
throw new
SemanticException(ConfigNodeMessages.CAN_ONLY_ALTER_DATATYPE_OF_FIELD_COLUMNS);
}
+ if (node.getPreAlteredColumns().containsKey(columnName)) {
+ final TSDataType currentType =
node.getPreAlteredColumns().get(columnName);
+ if (currentType == dataType) {
+ return;
+ }
+ throw new ColumnInAlterException(database.getFullPath(), tableName,
columnName);
+ }
+
if (!MetadataUtils.canAlter(columnSchema.getDataType(), dataType)) {
throw new SemanticException(
String.format(
@@ -1048,8 +1100,19 @@ public class ConfigMTree {
public void commitAlterColumnDataType(
PartialPath database, String tableName, String columnName, TSDataType
dataType)
throws MetadataException {
- final ConfigTableNode node = getTableNode(database, tableName);
- final TsTable table = getTable(database, tableName);
+ final IConfigMNode databaseNode =
getDatabaseNodeByDatabasePath(database).getAsMNode();
+ if (!databaseNode.hasChild(tableName)) {
+ return;
+ }
+ final IConfigMNode tableNode = databaseNode.getChild(tableName);
+ if (!(tableNode instanceof ConfigTableNode)) {
+ return;
+ }
+ final ConfigTableNode node = (ConfigTableNode) tableNode;
+ if (!Objects.equals(node.getPreAlteredColumns().get(columnName),
dataType)) {
+ return;
+ }
+ final TsTable table = node.getTable();
final TsTableColumnSchema columnSchema = table.getColumnSchema(columnName);
if (Objects.nonNull(columnSchema)) {
columnSchema.setDataType(dataType);
@@ -1058,20 +1121,68 @@ public class ConfigMTree {
fieldColumnSchema.setEncoding(
SchemaUtils.getDataTypeCompatibleEncoding(dataType,
fieldColumnSchema.getEncoding()));
}
+ }
+ node.removePreAlteredColumn(columnName);
+ }
+
+ public void rollbackPreAlterColumnDataType(
+ final PartialPath database,
+ final String tableName,
+ final String columnName,
+ final TSDataType dataType)
+ throws MetadataException {
+ final IConfigMNode databaseNode =
getDatabaseNodeByDatabasePath(database).getAsMNode();
+ if (!databaseNode.hasChild(tableName)) {
+ return;
+ }
+ final IConfigMNode tableNode = databaseNode.getChild(tableName);
+ if (!(tableNode instanceof ConfigTableNode)) {
+ return;
+ }
+ final ConfigTableNode node = (ConfigTableNode) tableNode;
+ if (Objects.equals(node.getPreAlteredColumns().get(columnName), dataType))
{
node.removePreAlteredColumn(columnName);
}
}
- public TsTable getUsingTableSchema(final PartialPath database, final String
tableName)
+ public TsTable getTableSchemaForDataNode(final PartialPath database, final
String tableName)
throws MetadataException {
- final ConfigTableNode node = getTableNode(database, tableName);
+ return getTableSchemaForDataNode(getTableNode(database, tableName));
+ }
+
+ private TsTable getTableSchemaForDataNode(final ConfigTableNode node) {
if (node.getPreDeletedColumns().isEmpty() &&
node.getPreAlteredColumns().isEmpty()) {
return node.getTable();
}
- final TsTable newTable = new TsTable(node.getTable());
- if (!node.getPreDeletedColumns().isEmpty()) {
- node.getPreDeletedColumns().forEach(newTable::removeColumnSchema);
+ // Cache reloads and later schema updates must not make a column writable
again while its
+ // deletion is still pending. DESC uses the complete schema separately.
+ final TsTable table = new TsTable(node.getTable());
+ node.getPreDeletedColumns().forEach(table::removeColumnSchema);
+ node.getPreAlteredColumns()
+ .forEach(
+ (columnName, dataType) -> {
+ final TsTableColumnSchema columnSchema =
table.getColumnSchema(columnName);
+ if (columnSchema == null) {
+ return;
+ }
+ columnSchema.setDataType(dataType);
+ if (columnSchema instanceof FieldColumnSchema) {
+ final FieldColumnSchema fieldColumnSchema =
(FieldColumnSchema) columnSchema;
+ fieldColumnSchema.setEncoding(
+ SchemaUtils.getDataTypeCompatibleEncoding(
+ dataType, fieldColumnSchema.getEncoding()));
+ }
+ });
+ return table;
+ }
+
+ public TsTable getTableSchemaForDesc(final PartialPath database, final
String tableName)
+ throws MetadataException {
+ final ConfigTableNode node = getTableNode(database, tableName);
+ if (node.getPreAlteredColumns().isEmpty()) {
+ return node.getTable();
}
+ final TsTable newTable = new TsTable(node.getTable());
if (!node.getPreAlteredColumns().isEmpty()) {
node.getPreAlteredColumns()
.forEach(
@@ -1110,6 +1221,30 @@ public class ConfigMTree {
return getTableNode(database, tableName).getTable();
}
+ private TsTable getTableForModification(
+ final PartialPath database, final String tableName, final String...
columnNames)
+ throws MetadataException {
+ final ConfigTableNode node = getTableNodeForModification(database,
tableName);
+ for (final String columnName : columnNames) {
+ if (node.getPreDeletedColumns().contains(columnName)) {
+ throw new ColumnInDeletionException(database.getFullPath(), tableName,
columnName);
+ }
+ if (node.getPreAlteredColumns().containsKey(columnName)) {
+ throw new ColumnInAlterException(database.getFullPath(), tableName,
columnName);
+ }
+ }
+ return node.getTable();
+ }
+
+ private ConfigTableNode getTableNodeForModification(
+ final PartialPath database, final String tableName) throws
MetadataException {
+ final ConfigTableNode node = getTableNode(database, tableName);
+ if (node.getStatus() == TableNodeStatus.PRE_DELETE) {
+ throw new TableInDeletionException(database.getFullPath(), tableName);
+ }
+ return node;
+ }
+
public Optional<Pair<TsTable, TableNodeStatus>> getTableAndStatusIfExists(
final PartialPath database, final String tableName) throws
MetadataException {
final IConfigMNode databaseNode =
getDatabaseNodeByDatabasePath(database).getAsMNode();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java
index 12491f24a48..dd63a1ecaf0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedure.java
@@ -24,9 +24,11 @@ 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.confignode.consensus.request.write.table.AlterColumnDataTypePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan;
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
+import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils;
import
org.apache.iotdb.confignode.procedure.state.schema.AlterTableColumnDataTypeState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -41,6 +43,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
+import java.util.Optional;
public class AlterTableColumnDataTypeProcedure
extends AbstractAlterOrDropTableProcedure<AlterTableColumnDataTypeState> {
@@ -156,14 +159,25 @@ public class AlterTableColumnDataTypeProcedure
new AlterColumnDataTypePlan(database, tableName, columnName,
dataType),
isGeneratedByPipe);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ // A consensus write may have been applied even when the client observed
a failure (for
+ // example, a timeout after the leader committed the entry). Continue
with release when the
+ // canonical CN schema already contains the requested type; rolling back
in that case would
+ // leave DataNodes on an older schema than the CN.
+ if (isColumnAlterCommitted(env)) {
+ setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE);
+ return;
+ }
setFailure(new ProcedureException(new
IoTDBException(status.getMessage(), status.getCode())));
+ return;
}
setNextState(AlterTableColumnDataTypeState.COMMIT_RELEASE);
}
@Override
protected boolean isRollbackSupported(final AlterTableColumnDataTypeState
state) {
- return false;
+ return state == AlterTableColumnDataTypeState.CHECK_AND_INVALIDATE_COLUMN
+ || state == AlterTableColumnDataTypeState.PRE_RELEASE
+ || state == AlterTableColumnDataTypeState.ALTER_TABLE_COLUMN_DATA_TYPE;
}
@Override
@@ -171,7 +185,67 @@ public class AlterTableColumnDataTypeProcedure
final ConfigNodeProcedureEnv configNodeProcedureEnv,
final AlterTableColumnDataTypeState alterTableColumnDataTypeState)
throws IOException, InterruptedException, ProcedureException {
- // Do nothing
+ // COMMIT_RELEASE is irreversible: the CN schema has already been
committed and must not be
+ // followed by a cache rollback if the procedure is aborted while
notifying DataNodes.
+ if (alterTableColumnDataTypeState ==
AlterTableColumnDataTypeState.COMMIT_RELEASE) {
+ return;
+ }
+ final Optional<TSDataType> pendingType =
getPreAlteredColumnType(configNodeProcedureEnv);
+ if (pendingType.isPresent() && pendingType.get() != dataType) {
+ // This rollback belongs to an older procedure. Leave a newer pre-alter
marker and its
+ // DataNode cache entry untouched.
+ return;
+ }
+ final boolean ownsPendingMarker = pendingType.isPresent();
+ if (!ownsPendingMarker &&
isColumnAlterCommittedForRollback(configNodeProcedureEnv)) {
+ return;
+ }
+ final TSStatus status =
+ SchemaUtils.executeInConsensusLayer(
+ new RollbackPreAlterColumnDataTypePlan(database, tableName,
columnName, dataType),
+ configNodeProcedureEnv,
+ LOGGER);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new ProcedureException(new IoTDBException(status.getMessage(),
status.getCode()));
+ }
+ if (alterTableColumnDataTypeState !=
AlterTableColumnDataTypeState.CHECK_AND_INVALIDATE_COLUMN
+ && table != null
+ && !getPreAlteredColumnType(configNodeProcedureEnv).isPresent()
+ && (ownsPendingMarker ||
!isColumnAlterCommittedForRollback(configNodeProcedureEnv))) {
+ rollbackPreRelease(configNodeProcedureEnv);
+ }
+ }
+
+ private Optional<TSDataType> getPreAlteredColumnType(final
ConfigNodeProcedureEnv env)
+ throws ProcedureException {
+ try {
+ return env.getConfigManager()
+ .getClusterSchemaManager()
+ .getPreAlteredColumnType(database, tableName, columnName);
+ } catch (final MetadataException e) {
+ throw new ProcedureException(e);
+ }
+ }
+
+ private boolean isColumnAlterCommittedForRollback(final
ConfigNodeProcedureEnv env)
+ throws ProcedureException {
+ try {
+ return env.getConfigManager()
+ .getClusterSchemaManager()
+ .isColumnAlterCommitted(database, tableName, columnName, dataType);
+ } catch (final MetadataException e) {
+ throw new ProcedureException(e);
+ }
+ }
+
+ private boolean isColumnAlterCommitted(final ConfigNodeProcedureEnv env) {
+ try {
+ return env.getConfigManager()
+ .getClusterSchemaManager()
+ .isColumnAlterCommitted(database, tableName, columnName, dataType);
+ } catch (final MetadataException e) {
+ return false;
+ }
}
@Override
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 f2379ae1256..8c6596cfe9e 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
@@ -22,6 +22,8 @@ package
org.apache.iotdb.confignode.procedure.impl.schema.table;
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.exception.table.TableInDeletionException;
+import org.apache.iotdb.commons.schema.table.TableNodeStatus;
import org.apache.iotdb.commons.schema.table.TsTable;
import
org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
@@ -40,6 +42,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +52,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import static org.apache.iotdb.rpc.TSStatusCode.TABLE_ALREADY_EXISTS;
@@ -117,10 +121,14 @@ public class CreateTableProcedure
protected void checkTableExistence(final ConfigNodeProcedureEnv env) {
try {
- if (env.getConfigManager()
- .getClusterSchemaManager()
- .getTableIfExists(database, table.getTableName())
- .isPresent()) {
+ final Optional<Pair<TsTable, TableNodeStatus>> existingTable =
+ env.getConfigManager()
+ .getClusterSchemaManager()
+ .getTableAndStatusIfExists(database, table.getTableName());
+ if (existingTable.isPresent()) {
+ if (existingTable.get().getRight() == TableNodeStatus.PRE_DELETE) {
+ throw new TableInDeletionException(database, table.getTableName());
+ }
setFailure(
new ProcedureException(
new IoTDBException(
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 0e858589803..1b9f4d55029 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
@@ -147,6 +147,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTableP
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RenameTablePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackCreateTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.RollbackPreAlterColumnDataTypePlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableCommentPlan;
import
org.apache.iotdb.confignode.consensus.request.write.table.SetTablePropertiesPlan;
@@ -1716,6 +1717,20 @@ public class ConfigPhysicalPlanSerDeTest {
alterColumnDataTypePlan.getNewType(),
alterColumnDataTypePlan1.getNewType());
}
+ @Test
+ public void RollbackPreAlterTableColumnDataTypePlanTest() throws IOException
{
+ final RollbackPreAlterColumnDataTypePlan rollbackPlan =
+ new RollbackPreAlterColumnDataTypePlan("database1", "table1", "field",
TSDataType.FLOAT);
+ final RollbackPreAlterColumnDataTypePlan rollbackPlan1 =
+ (RollbackPreAlterColumnDataTypePlan)
+
ConfigPhysicalPlan.Factory.create(rollbackPlan.serializeToByteBuffer());
+ Assert.assertEquals(rollbackPlan.getDatabase(),
rollbackPlan1.getDatabase());
+ Assert.assertEquals(rollbackPlan.getTableName(),
rollbackPlan1.getTableName());
+ Assert.assertEquals(rollbackPlan.getColumnName(),
rollbackPlan1.getColumnName());
+ Assert.assertEquals(rollbackPlan.getType(), rollbackPlan1.getType());
+ Assert.assertEquals(rollbackPlan.getNewType(), rollbackPlan1.getNewType());
+ }
+
@Test
public void AlterTableColumnDataTypePlanTest() throws IOException {
final AlterColumnDataTypePlan alterColumnDataTypePlan =
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java
index c2519dcbfae..318c957b825 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTreeTest.java
@@ -49,6 +49,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -360,6 +361,10 @@ public class ConfigMTreeTest {
root.preCreateTable(pathList[i], table);
root.commitCreateTable(pathList[i], tableName);
+ if (i == 0) {
+ Assert.assertTrue(root.preDeleteColumn(pathList[i], tableName, "Attr",
false));
+ root.preAlterColumnDataType(pathList[i], tableName, "Measurement",
TSDataType.STRING);
+ }
}
final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
@@ -402,8 +407,21 @@ public class ConfigMTreeTest {
final TsTable table = tables.get(0);
assertEquals("table" + i, table.getTableName());
assertEquals(1, table.getTagNum());
- // currently, only construct the TsTable would not carry the time column
- assertEquals(3, table.getColumnNum());
+ // These TsTables have no time column; the first table also hides its
pre-deleted attribute.
+ assertEquals(i == 0 ? 2 : 3, table.getColumnNum());
+ final ConfigMTree.TableSchemaDetails details =
+ newTree.getTableSchemaDetails(pathList[i], table.getTableName());
+ if (i == 0) {
+ assertEquals(Collections.singleton("Attr"), details.preDeletedColumns);
+ assertEquals(TSDataType.STRING,
details.preAlteredColumns.get("Measurement"));
+ assertEquals(TSDataType.DOUBLE,
details.table.getColumnSchema("Measurement").getDataType());
+ Assert.assertNotNull(details.table.getColumnSchema("Attr"));
+ assertEquals(TSDataType.STRING,
table.getColumnSchema("Measurement").getDataType());
+ Assert.assertNull(table.getColumnSchema("Attr"));
+ } else {
+ assertTrue(details.preDeletedColumns.isEmpty());
+ assertTrue(details.preAlteredColumns.isEmpty());
+ }
}
}
@@ -434,7 +452,7 @@ public class ConfigMTreeTest {
SchemaUtils.getDataTypeCompatibleEncoding(TSDataType.STRING,
TSEncoding.GORILLA);
Assert.assertNotEquals(TSEncoding.GORILLA, expectedEncoding);
- final TsTable preAlteredTable = root.getUsingTableSchema(database,
table.getTableName());
+ final TsTable preAlteredTable = root.getTableSchemaForDesc(database,
table.getTableName());
final FieldColumnSchema preAlteredField =
(FieldColumnSchema) preAlteredTable.getColumnSchema("measurement");
Assert.assertEquals(TSDataType.STRING, preAlteredField.getDataType());
@@ -443,7 +461,7 @@ public class ConfigMTreeTest {
root.commitAlterColumnDataType(
database, table.getTableName(), "measurement", TSDataType.STRING);
- final TsTable committedTable = root.getUsingTableSchema(database,
table.getTableName());
+ final TsTable committedTable = root.getTableSchemaForDesc(database,
table.getTableName());
final FieldColumnSchema committedField =
(FieldColumnSchema) committedTable.getColumnSchema("measurement");
Assert.assertEquals(TSDataType.STRING, committedField.getDataType());
@@ -453,6 +471,57 @@ public class ConfigMTreeTest {
root.getTableSchemaDetails(database,
table.getTableName()).preAlteredColumns.isEmpty());
}
+ @Test
+ public void testRollbackPreAlterColumnDataTypeOnlyClearsMatchingRequest()
throws Exception {
+ root = new ConfigMTree(true);
+
+ final PartialPath database = new PartialPath("root.sg");
+ root.setStorageGroup(database);
+ final IDatabaseMNode<IConfigMNode> databaseNode =
root.getDatabaseNodeByDatabasePath(database);
+ databaseNode
+ .getAsMNode()
+ .getDatabaseSchema()
+ .setName(PathUtils.unQualifyDatabaseName(database.getFullPath()));
+ databaseNode.getAsMNode().getDatabaseSchema().setIsTableModel(true);
+
+ final TsTable table = new TsTable("table1");
+ table.addColumnSchema(new TagColumnSchema("id", TSDataType.STRING));
+ table.addColumnSchema(
+ new FieldColumnSchema(
+ "measurement", TSDataType.DOUBLE, TSEncoding.GORILLA,
CompressionType.SNAPPY));
+ root.preCreateTable(database, table);
+ root.commitCreateTable(database, table.getTableName());
+
+ root.preAlterColumnDataType(database, table.getTableName(), "measurement",
TSDataType.STRING);
+ Assert.assertEquals(
+ TSDataType.STRING,
+ root.getTableSchemaForDataNode(database, table.getTableName())
+ .getColumnSchema("measurement")
+ .getDataType());
+
+ // A stale rollback must not clear a newer request for a different target
type.
+ root.rollbackPreAlterColumnDataType(
+ database, table.getTableName(), "measurement", TSDataType.FLOAT);
+ Assert.assertEquals(
+ TSDataType.STRING,
+ root.getTableSchemaDetails(database, table.getTableName())
+ .preAlteredColumns
+ .get("measurement"));
+
+ root.rollbackPreAlterColumnDataType(
+ database, table.getTableName(), "measurement", TSDataType.STRING);
+ Assert.assertTrue(
+ root.getTableSchemaDetails(database,
table.getTableName()).preAlteredColumns.isEmpty());
+ // A delayed commit from the failed procedure must not apply after the
marker was rolled back.
+ root.commitAlterColumnDataType(
+ database, table.getTableName(), "measurement", TSDataType.STRING);
+ Assert.assertEquals(
+ TSDataType.DOUBLE,
+ root.getTableSchemaForDataNode(database, table.getTableName())
+ .getColumnSchema("measurement")
+ .getDataType());
+ }
+
@Test
public void testSetTemplate() throws MetadataException {
root.setStorageGroup(new PartialPath("root.a"));
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java
new file mode 100644
index 00000000000..86b74ec00b6
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/schema/TablePreDeleteTest.java
@@ -0,0 +1,368 @@
+/*
+ * 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;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.exception.table.ColumnInAlterException;
+import org.apache.iotdb.commons.exception.table.ColumnInDeletionException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
+import org.apache.iotdb.commons.schema.table.TableNodeStatus;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
+import org.apache.iotdb.commons.schema.table.column.AttributeColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.read.table.DescTablePlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.FetchTablePlan;
+import org.apache.iotdb.confignode.consensus.request.read.table.ShowTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.AddTableColumnPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.AlterColumnDataTypePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.CommitCreateTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.CommitDeleteColumnPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.CommitDeleteTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.PreAlterColumnDataTypePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.PreCreateTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteColumnPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.PreDeleteTablePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.table.SetTableColumnCommentPlan;
+import org.apache.iotdb.confignode.manager.IManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaQuotaStatistics;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
+import org.apache.iotdb.confignode.rpc.thrift.TDescTableResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp;
+import org.apache.iotdb.confignode.rpc.thrift.TTableInfo;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+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.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+public class TablePreDeleteTest {
+ private static final String DATABASE = "root.pre_delete_test";
+ private static final String TABLE = "table1";
+ private ClusterSchemaInfo schemaInfo;
+ private ClusterSchemaManager schemaManager;
+
+ @Before
+ public void setUp() throws Exception {
+ schemaInfo = new ClusterSchemaInfo();
+ schemaManager =
+ new ClusterSchemaManager(
+ Mockito.mock(IManager.class),
+ schemaInfo,
+ Mockito.mock(ClusterSchemaQuotaStatistics.class));
+ assertSuccess(
+ schemaInfo.createDatabase(
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase,
+ new TDatabaseSchema(DATABASE).setIsTableModel(true))));
+ createTable(TABLE);
+ }
+
+ @After
+ public void tearDown() {
+ schemaInfo.clear();
+ }
+
+ @Test
+ public void testShowTablesIncludesPreDeleteButNotPreCreate() {
+ createTable("using_table");
+ assertSuccess(
+ schemaInfo.preCreateTable(new PreCreateTablePlan(DATABASE, new
TsTable("creating"))));
+ assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE,
TABLE)));
+
+ final TShowTableResp basic =
+ schemaInfo.showTables(new ShowTablePlan(DATABASE,
false)).convertToTShowTableResp();
+ assertSuccess(basic.getStatus());
+ assertEquals(
+ Arrays.asList(TABLE, "using_table"),
+ basic.getTableInfoList().stream()
+ .map(TTableInfo::getTableName)
+ .sorted()
+ .collect(Collectors.toList()));
+
+ final TShowTableResp details =
+ schemaInfo.showTables(new ShowTablePlan(DATABASE,
true)).convertToTShowTableResp();
+ final Map<String, Integer> states =
+ details.getTableInfoList().stream()
+ .collect(Collectors.toMap(TTableInfo::getTableName,
TTableInfo::getState));
+ assertEquals(Integer.valueOf(TableNodeStatus.PRE_DELETE.ordinal()),
states.get(TABLE));
+ assertEquals(Integer.valueOf(TableNodeStatus.PRE_CREATE.ordinal()),
states.get("creating"));
+ assertEquals(Integer.valueOf(TableNodeStatus.USING.ordinal()),
states.get("using_table"));
+
+ assertSuccess(schemaInfo.dropTable(new CommitDeleteTablePlan(DATABASE,
TABLE)));
+ assertEquals(
+ Collections.singletonList("using_table"),
+ schemaInfo
+ .showTables(new ShowTablePlan(DATABASE, false))
+ .convertToTShowTableResp()
+ .getTableInfoList()
+ .stream()
+ .map(TTableInfo::getTableName)
+ .collect(Collectors.toList()));
+ }
+
+ @Test
+ public void testDescRetainsPreDeletedColumnsAndAlteredTypes() {
+ assertSuccess(schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE,
TABLE, "field")));
+ assertSuccess(
+ schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE,
"attribute")));
+ assertSuccess(
+ schemaInfo.preAlterColumnDataType(
+ new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live",
TSDataType.INT64)));
+
+ TDescTableResp basic = describe(false);
+ TsTable table =
TsTableInternalRPCUtil.deserializeSingleTsTable(basic.getTableInfo());
+ assertNotNull(table.getColumnSchema("field"));
+ assertNotNull(table.getColumnSchema("attribute"));
+ assertEquals(TSDataType.INT64,
table.getColumnSchema("live").getDataType());
+ assertFalse(basic.isSetPreDeletedColumns());
+
+ assertEquals(
+ TSDataType.INT64,
+
schemaInfo.getAllUsingTables().get(DATABASE).get(0).getColumnSchema("live").getDataType());
+
+ final TDescTableResp details = describe(true);
+
assertTrue(details.getPreDeletedColumns().containsAll(Arrays.asList("field",
"attribute")));
+ assertEquals(
+ Byte.valueOf(TSDataType.INT64.serialize()),
details.getPreAlteredColumns().get("live"));
+
+ assertSuccess(
+ schemaInfo.commitDeleteColumn(new CommitDeleteColumnPlan(DATABASE,
TABLE, "field")));
+ table =
TsTableInternalRPCUtil.deserializeSingleTsTable(describe(false).getTableInfo());
+ assertNull(table.getColumnSchema("field"));
+
+ assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE,
TABLE)));
+ assertNotNull(
+
TsTableInternalRPCUtil.deserializeSingleTsTable(describe(false).getTableInfo())
+ .getColumnSchema("attribute"));
+ }
+
+ @Test
+ public void testColumnExtensionRejectsPreDeletedNames() throws Exception {
+ for (final String column : Arrays.asList("field", "attribute")) {
+ assertSuccess(schemaInfo.preDeleteColumn(new
PreDeleteColumnPlan(DATABASE, TABLE, column)));
+ final List<TsTableColumnSchema> columns =
+ new ArrayList<>(Arrays.asList(field("new_field"), field(column)));
+ final ColumnInDeletionException exception =
+ assertThrows(
+ ColumnInDeletionException.class,
+ () ->
+ schemaManager.tableColumnCheckForColumnExtension(
+ DATABASE, TABLE, columns, false));
+ assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(),
exception.getErrorCode());
+ assertEquals(
+ new ColumnInDeletionException(DATABASE, TABLE, column).getMessage(),
+ exception.getMessage());
+ assertEquals(2, columns.size());
+ assertNull(
+ schemaInfo
+ .getTsTableIfExists(DATABASE, TABLE)
+ .get()
+ .getLeft()
+ .getColumnSchema("new_field"));
+ }
+ assertEquals(
+ TSStatusCode.COLUMN_ALREADY_EXISTS.getStatusCode(),
+ schemaManager
+ .tableColumnCheckForColumnExtension(
+ DATABASE, TABLE, new
ArrayList<>(Collections.singletonList(field("live"))), false)
+ .getLeft()
+ .getCode());
+
+ assertSuccess(
+ schemaInfo.commitDeleteColumn(new CommitDeleteColumnPlan(DATABASE,
TABLE, "field")));
+ assertSuccess(
+ schemaManager
+ .tableColumnCheckForColumnExtension(
+ DATABASE, TABLE, new
ArrayList<>(Collections.singletonList(field("field"))), false)
+ .getLeft());
+ }
+
+ @Test
+ public void testPreAlterRejectsConflictingColumnOperations() throws
Exception {
+ assertSuccess(
+ schemaInfo.preAlterColumnDataType(
+ new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live",
TSDataType.INT64)));
+ // Retrying the same target type is allowed so a stuck procedure can be
resumed.
+ assertSuccess(
+ schemaInfo.preAlterColumnDataType(
+ new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live",
TSDataType.INT64)));
+
+ final TSStatus secondAlter =
+ schemaInfo.preAlterColumnDataType(
+ new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live",
TSDataType.FLOAT));
+ assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(),
secondAlter.getCode());
+ assertEquals(
+ new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(),
secondAlter.getMessage());
+
+ final TSStatus delete =
+ schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE,
"live"));
+ assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(),
delete.getCode());
+ assertEquals(
+ new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(),
delete.getMessage());
+
+ assertThrows(
+ ColumnInAlterException.class,
+ () ->
+ schemaManager.tableColumnCheckForColumnExtension(
+ DATABASE, TABLE, new
ArrayList<>(Collections.singletonList(field("live"))), false));
+ assertEquals(
+ new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(),
+ schemaInfo
+ .addTableColumn(
+ new AddTableColumnPlan(
+ DATABASE, TABLE, Collections.singletonList(field("live")),
false))
+ .getMessage());
+ assertEquals(
+ new ColumnInAlterException(DATABASE, TABLE, "live").getMessage(),
+ schemaInfo
+ .setTableColumnComment(
+ new SetTableColumnCommentPlan(DATABASE, TABLE, "live",
"comment"))
+ .getMessage());
+ assertThrows(
+ ColumnInAlterException.class,
+ () ->
+ schemaManager.tableColumnCheckForColumnRenaming(
+ DATABASE, TABLE, "live", "renamed", false));
+ }
+
+ @Test
+ public void
testSameTypePreAlterIsNotReportedAsCommittedUntilMarkerIsCleared() throws
Exception {
+ assertSuccess(
+ schemaInfo.preAlterColumnDataType(
+ new PreAlterColumnDataTypePlan(DATABASE, TABLE, "live",
TSDataType.INT32)));
+ assertFalse(schemaInfo.isColumnAlterCommitted(DATABASE, TABLE, "live",
TSDataType.INT32));
+
+ assertSuccess(
+ schemaInfo.commitAlterColumnDataType(
+ new AlterColumnDataTypePlan(DATABASE, TABLE, "live",
TSDataType.INT32)));
+ assertTrue(schemaInfo.isColumnAlterCommitted(DATABASE, TABLE, "live",
TSDataType.INT32));
+ }
+
+ @Test
+ public void testPreDeletedTableRejectsCreationAndColumnExtension() {
+ assertSuccess(schemaInfo.preDeleteTable(new PreDeleteTablePlan(DATABASE,
TABLE)));
+ final TSStatus create =
+ schemaInfo.preCreateTable(new PreCreateTablePlan(DATABASE, new
TsTable(TABLE)));
+ assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(),
create.getCode());
+ assertEquals(new TableInDeletionException(DATABASE, TABLE).getMessage(),
create.getMessage());
+ final TableInDeletionException exception =
+ assertThrows(
+ TableInDeletionException.class,
+ () ->
+ schemaManager.tableColumnCheckForColumnExtension(
+ DATABASE,
+ TABLE,
+ new
ArrayList<>(Collections.singletonList(field("new_field"))),
+ false));
+ assertEquals(create.getMessage(), exception.getMessage());
+ }
+
+ @Test
+ public void testDataNodeSchemasExcludePreDeletedColumns() throws Exception {
+ assertSuccess(schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE,
TABLE, "field")));
+ assertSuccess(
+ schemaInfo.preDeleteColumn(new PreDeleteColumnPlan(DATABASE, TABLE,
"attribute")));
+
+ assertOnlyLiveColumns(schemaInfo.getAllUsingTables().get(DATABASE).get(0));
+ final TsTable fetched =
+ TsTableInternalRPCUtil.deserializeTsTableFetchResult(
+ schemaInfo
+ .fetchTables(
+ new FetchTablePlan(
+ Collections.singletonMap(DATABASE,
Collections.singleton(TABLE)),
+ Collections.singleton(TableNodeStatus.USING)))
+ .convertToTFetchTableResp()
+ .getTableInfoMap())
+ .get(DATABASE)
+ .get(TABLE);
+ assertOnlyLiveColumns(fetched);
+
+ final TsTable expanded =
+ schemaManager
+ .tableColumnCheckForColumnExtension(
+ DATABASE,
+ TABLE,
+ new ArrayList<>(Collections.singletonList(field("new_field"))),
+ false)
+ .getRight();
+ assertOnlyLiveColumns(expanded);
+ assertNotNull(expanded.getColumnSchema("new_field"));
+
+ // Preparing a cache snapshot must not change the complete schema used by
DESC.
+ final TsTable described =
+
TsTableInternalRPCUtil.deserializeSingleTsTable(describe(false).getTableInfo());
+ assertNotNull(described.getColumnSchema("field"));
+ assertNotNull(described.getColumnSchema("attribute"));
+ }
+
+ private static void assertOnlyLiveColumns(final TsTable table) {
+ assertNotNull(table.getColumnSchema("live"));
+ assertNull(table.getColumnSchema("field"));
+ assertNull(table.getColumnSchema("attribute"));
+ }
+
+ private TDescTableResp describe(final boolean details) {
+ final TDescTableResp resp =
+ schemaInfo.descTable(new DescTablePlan(DATABASE, TABLE,
details)).convertToTDescTableResp();
+ assertSuccess(resp.getStatus());
+ return resp;
+ }
+
+ private void createTable(final String name) {
+ final TsTable table = new TsTable(name);
+ table.addColumnSchema(field("field"));
+ table.addColumnSchema(field("live"));
+ table.addColumnSchema(new AttributeColumnSchema("attribute",
TSDataType.STRING));
+ assertSuccess(schemaInfo.preCreateTable(new PreCreateTablePlan(DATABASE,
table)));
+ assertSuccess(schemaInfo.commitCreateTable(new
CommitCreateTablePlan(DATABASE, name)));
+ }
+
+ private static FieldColumnSchema field(final String name) {
+ return new FieldColumnSchema(name, TSDataType.INT32, TSEncoding.RLE,
CompressionType.LZ4);
+ }
+
+ private static void assertSuccess(final TSStatus status) {
+ assertEquals(
+ status.getMessage(), TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
+ }
+}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java
new file mode 100644
index 00000000000..5634abbe6bf
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/table/AlterTableColumnDataTypeProcedureTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.TSStatus;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import
org.apache.iotdb.confignode.procedure.state.schema.AlterTableColumnDataTypeState;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class AlterTableColumnDataTypeProcedureTest {
+
+ @Test
+ public void commitReleaseIsNeverRolledBack() throws Exception {
+ final AlterTableColumnDataTypeProcedure procedure =
+ new AlterTableColumnDataTypeProcedure("database", "table", "query",
"value", null, false);
+ procedure.table = new TsTable("table");
+
+ // The commit state may be present on the rollback stack when an abort
races with cache
+ // notification. It must not issue a rollback plan or touch DataNode
caches.
+ procedure.rollbackState(null,
AlterTableColumnDataTypeState.COMMIT_RELEASE);
+ }
+
+ @Test
+ public void rollbackClearsCnMarkerAfterProcedureRestartBeforeTableSnapshot()
throws Exception {
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ final ClusterSchemaManager schemaManager =
Mockito.mock(ClusterSchemaManager.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+
Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager);
+ Mockito.when(consensusManager.write(Mockito.any()))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ Mockito.when(
+ schemaManager.isColumnAlterCommitted("database", "table", "value",
TSDataType.INT64))
+ .thenReturn(false);
+
+ final AlterTableColumnDataTypeProcedure procedure =
+ new AlterTableColumnDataTypeProcedure(
+ "database", "table", "query", "value", TSDataType.INT64, false);
+
+ // The table snapshot is intentionally null, as it can be after a restart
that happened after
+ // the pre-alter consensus entry was applied but before the procedure
persisted its snapshot.
+ procedure.rollbackState(env,
AlterTableColumnDataTypeState.CHECK_AND_INVALIDATE_COLUMN);
+
+ Mockito.verify(consensusManager).write(Mockito.any());
+ }
+
+ @Test
+ public void sameTypePreAlterRollbackStillCleansDataNodeCache() throws
Exception {
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ final ClusterSchemaManager schemaManager =
Mockito.mock(ClusterSchemaManager.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+
Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager);
+
+ final AtomicBoolean markerCleared = new AtomicBoolean(false);
+ Mockito.when(schemaManager.getPreAlteredColumnType("database", "table",
"value"))
+ .thenAnswer(
+ invocation -> markerCleared.get() ? Optional.empty() :
Optional.of(TSDataType.INT64));
+ Mockito.when(
+ schemaManager.isColumnAlterCommitted("database", "table", "value",
TSDataType.INT64))
+ .thenAnswer(invocation -> markerCleared.get());
+ Mockito.when(consensusManager.write(Mockito.any()))
+ .thenAnswer(
+ invocation -> {
+ markerCleared.set(true);
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ });
+
+ final TestAlterTableColumnDataTypeProcedure procedure =
+ new TestAlterTableColumnDataTypeProcedure();
+ procedure.table = new TsTable("table");
+ procedure.rollbackState(env, AlterTableColumnDataTypeState.PRE_RELEASE);
+
+ Assert.assertTrue(procedure.dataNodeRollbackCalled);
+ }
+
+ @Test
+ public void staleRollbackDoesNotClearNewerPreAlter() throws Exception {
+ final ConsensusManager consensusManager =
Mockito.mock(ConsensusManager.class);
+ final ClusterSchemaManager schemaManager =
Mockito.mock(ClusterSchemaManager.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+
Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager);
+ Mockito.when(schemaManager.getPreAlteredColumnType("database", "table",
"value"))
+ .thenReturn(Optional.of(TSDataType.FLOAT));
+
+ final TestAlterTableColumnDataTypeProcedure procedure =
+ new TestAlterTableColumnDataTypeProcedure();
+ procedure.table = new TsTable("table");
+ procedure.rollbackState(env, AlterTableColumnDataTypeState.PRE_RELEASE);
+
+ Mockito.verify(consensusManager, Mockito.never()).write(Mockito.any());
+ Assert.assertFalse(procedure.dataNodeRollbackCalled);
+ }
+
+ @Test
+ public void consensusFailureAfterCommitContinuesToCommitRelease() throws
Exception {
+ final ClusterSchemaManager schemaManager =
Mockito.mock(ClusterSchemaManager.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager);
+
+ Mockito.when(schemaManager.executePlan(Mockito.any(), Mockito.eq(false)))
+ .thenReturn(new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
+ Mockito.when(
+ schemaManager.isColumnAlterCommitted("database", "table", "value",
TSDataType.INT64))
+ .thenReturn(true);
+
+ final AlterTableColumnDataTypeProcedure procedure =
+ new AlterTableColumnDataTypeProcedure(
+ "database", "table", "query", "value", TSDataType.INT64, false);
+ final Method alterColumnDataType =
+ AlterTableColumnDataTypeProcedure.class.getDeclaredMethod(
+ "alterColumnDataType", ConfigNodeProcedureEnv.class);
+ alterColumnDataType.setAccessible(true);
+ alterColumnDataType.invoke(procedure, env);
+
+ Assert.assertFalse(procedure.isFailed());
+ }
+
+ private static class TestAlterTableColumnDataTypeProcedure
+ extends AlterTableColumnDataTypeProcedure {
+ private boolean dataNodeRollbackCalled;
+
+ private TestAlterTableColumnDataTypeProcedure() {
+ super("database", "table", "query", "value", TSDataType.INT64, false);
+ }
+
+ @Override
+ protected void rollbackPreRelease(final ConfigNodeProcedureEnv env) {
+ dataNodeRollbackCalled = true;
+ }
+ }
+}
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
index fab1bfb2660..0b9a118fd1a 100644
---
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
@@ -20,24 +20,63 @@
package org.apache.iotdb.confignode.procedure.impl.schema.table;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
+import org.apache.iotdb.commons.schema.table.TableNodeStatus;
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.FieldColumnSchema;
import org.apache.iotdb.commons.schema.table.column.TagColumnSchema;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.utils.Pair;
import org.junit.Assert;
import org.junit.Test;
+import org.mockito.Mockito;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Optional;
public class CreateTableProcedureTest {
+ @Test
+ public void testPreDeletedTableIsNotReportedAsAlreadyExisting() throws
Exception {
+ final ConfigNodeProcedureEnv env =
Mockito.mock(ConfigNodeProcedureEnv.class);
+ final ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ final ClusterSchemaManager schemaManager =
Mockito.mock(ClusterSchemaManager.class);
+ Mockito.when(env.getConfigManager()).thenReturn(configManager);
+
Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager);
+ final TsTable table = new TsTable("table1");
+ Mockito.when(schemaManager.getTableAndStatusIfExists("database1",
"table1"))
+ .thenReturn(Optional.of(new Pair<>(table,
TableNodeStatus.PRE_DELETE)));
+
+ final CreateTableProcedure procedure = new
CreateTableProcedure("database1", table, false);
+ procedure.checkTableExistence(env);
+
+ Assert.assertTrue(procedure.isFailed());
+ Assert.assertTrue(procedure.getException().getCause() instanceof
TableInDeletionException);
+ Assert.assertEquals(
+ TSStatusCode.SEMANTIC_ERROR.getStatusCode(),
+ ((IoTDBException) procedure.getException().getCause()).getErrorCode());
+
+ Mockito.when(schemaManager.getTableAndStatusIfExists("database1",
"table1"))
+ .thenReturn(Optional.of(new Pair<>(table, TableNodeStatus.USING)));
+ final CreateTableProcedure duplicate = new
CreateTableProcedure("database1", table, false);
+ duplicate.checkTableExistence(env);
+ Assert.assertEquals(
+ TSStatusCode.TABLE_ALREADY_EXISTS.getStatusCode(),
+ ((IoTDBException) duplicate.getException().getCause()).getErrorCode());
+ }
+
@Test
public void serializeDeserializeTest() throws IllegalPathException,
IOException {
final TsTable table = new TsTable("table1");
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
index dd1c4af94f6..ef703af3f0e 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
@@ -584,8 +584,6 @@ public final class DataNodeSchemaMessages {
public static final String UPDATE_TABLE_BY_FETCH_WITH_DETAIL =
"Update table {}.{} by table fetch, {}";
public static final String UPDATE_TABLE_BY_FETCH = "Update table {}.{} by
table fetch.";
- public static final String THE_TABLE_IS_IN_PRE_DELETE_STATE =
- "The table %s.%s is in the pre-delete state. Please wait a few seconds.
If the table is still in this state, please drop it again.";
public static final String COMPARE_TABLE_ADDED = "Added table: ";
public static final String COMPARE_TABLE_REMOVED = "Removed table: ";
public static final String COMPARE_TABLE_NAME = "Table name: ";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
index 411044b14af..f1ceaaa0470 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
@@ -578,8 +578,6 @@ public final class DataNodeSchemaMessages {
"尝试获取信号量以从 ConfigNode 获取表时被中断，已忽略。";
public static final String UPDATE_TABLE_BY_FETCH_WITH_DETAIL = "获取表 {}.{}
信息, {}";
public static final String UPDATE_TABLE_BY_FETCH = "通过表拉取更新表 {}.{}";
- public static final String THE_TABLE_IS_IN_PRE_DELETE_STATE =
- "表 %s.%s 处于预删除的状态，请稍等，如之后重试还是此状态，请输入sql再次删除";
public static final String COMPARE_TABLE_ADDED = "新增表：";
public static final String COMPARE_TABLE_REMOVED = "已移除表：";
public static final String COMPARE_TABLE_NAME = "表名：";
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 7a3580fe66b..d7dd2abcb1d 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
@@ -4830,6 +4830,20 @@ public class ClusterConfigTaskExecutor implements
IConfigTaskExecutor {
return future;
}
+ public Set<String> getPreDeletedColumns(final String database, final String
tableName) {
+ try (final ConfigNodeClient configNodeClient =
+
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ final TDescTableResp resp = configNodeClient.describeTable(database,
tableName, true);
+ if (resp.getStatus().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(
+ getTableErrorMessage(resp.getStatus(), database),
resp.getStatus().getCode());
+ }
+ return resp.isSetPreDeletedColumns() ? resp.getPreDeletedColumns() :
Collections.emptySet();
+ } catch (final ClientManagerException | TException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
public SettableFuture<ConfigTaskResult> describeTable(
final String database,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java
index 6a08427d82b..aeaf82e54f0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidator.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.exception.table.ColumnInDeletionException;
import org.apache.iotdb.commons.i18n.QueryMessages;
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema;
import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
@@ -71,6 +72,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -83,11 +85,14 @@ public class TableHeaderSchemaValidator {
private static final Logger LOGGER =
LoggerFactory.getLogger(TableHeaderSchemaValidator.class);
- private final ClusterConfigTaskExecutor configTaskExecutor =
- ClusterConfigTaskExecutor.getInstance();
+ private final ClusterConfigTaskExecutor configTaskExecutor;
private TableHeaderSchemaValidator() {
- // do nothing
+ this(ClusterConfigTaskExecutor.getInstance());
+ }
+
+ TableHeaderSchemaValidator(final ClusterConfigTaskExecutor
configTaskExecutor) {
+ this.configTaskExecutor = configTaskExecutor;
}
private static class TableHeaderSchemaValidatorHolder {
@@ -213,6 +218,10 @@ public class TableHeaderSchemaValidator {
boolean refreshed = false;
boolean noField = true;
+ // Keep this cache scoped to the validation request. ConfigNode does not
notify DataNode when
+ // DROP COLUMN commits, so a cross-request cache could reject a recreated
column as still
+ // being deleted.
+ Set<String> preDeletedColumns = null;
for (final ColumnSchema columnSchema : inputColumnList) {
TsTableColumnSchema existingColumn =
table.getColumnSchema(columnSchema.getName());
if (Objects.isNull(existingColumn)) {
@@ -225,6 +234,12 @@ public class TableHeaderSchemaValidator {
existingColumn = table.getColumnSchema(columnSchema.getName());
}
if (Objects.isNull(existingColumn)) {
+ if (preDeletedColumns == null) {
+ preDeletedColumns =
+ configTaskExecutor.getPreDeletedColumns(database,
tableSchema.getTableName());
+ }
+ checkColumnNotPreDeleted(
+ database, tableSchema.getTableName(), columnSchema.getName(),
preDeletedColumns);
// check arguments for column auto creation
if (columnSchema.getColumnCategory() == null) {
throw new SemanticException(
@@ -397,6 +412,10 @@ public class TableHeaderSchemaValidator {
boolean refreshed = false;
boolean noField = true;
boolean hasAttribute = false;
+ // Keep this cache scoped to the validation request. ConfigNode does not
notify DataNode when
+ // DROP COLUMN commits, so a cross-request cache could reject a recreated
column as still
+ // being deleted.
+ Set<String> preDeletedColumns = null;
// Track TAG column measurement indices for batch processing after
validation loop
// LinkedHashMap maintains insertion order, key is column name, value is
measurement index
@@ -432,6 +451,12 @@ public class TableHeaderSchemaValidator {
}
if (Objects.isNull(existingColumn)) {
+ if (preDeletedColumns == null) {
+ preDeletedColumns =
+ configTaskExecutor.getPreDeletedColumns(database,
measurementInfo.getTableName());
+ }
+ checkColumnNotPreDeleted(
+ database, measurementInfo.getTableName(), measurementName,
preDeletedColumns);
// Check arguments for column auto creation
if (category == null) {
throw new SemanticException(
@@ -539,6 +564,16 @@ public class TableHeaderSchemaValidator {
}
}
+ private static void checkColumnNotPreDeleted(
+ final String database,
+ final String tableName,
+ final String columnName,
+ final Set<String> preDeletedColumns) {
+ if (preDeletedColumns.contains(columnName)) {
+ throw new SemanticException(new ColumnInDeletionException(database,
tableName, columnName));
+ }
+ }
+
private void autoCreateTableFromMeasurementInfo(
final MPPQueryContext context,
final String database,
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 23daf268206..39520259da1 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
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.ConfigRegionId;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
import org.apache.iotdb.commons.schema.table.NonCommittableTsTable;
import org.apache.iotdb.commons.schema.table.PreDeleteTsTable;
import org.apache.iotdb.commons.schema.table.TableNodeStatus;
@@ -87,6 +88,13 @@ public class DataNodeTableCache implements ITableCache {
private final Map<String, Map<String, Pair<TsTable, Long>>> specialStatusMap
=
new ConcurrentHashMap<>();
+ /**
+ * The cache entry replaced by the latest pre-update. It lets a rollback
restore a schema that may
+ * already have been promoted from {@code specialStatusMap} by a concurrent
fetch. A {@link
+ * NonCommittableTsTable} value means that the previous schema is
unavailable after a restart.
+ */
+ private final Map<String, Map<String, TsTable>> previousTableMap = new
ConcurrentHashMap<>();
+
private final ReentrantReadWriteLock readWriteLock = new
ReentrantReadWriteLock();
private final Semaphore fetchTableSemaphore =
new Semaphore(
@@ -119,6 +127,7 @@ public class DataNodeTableCache implements ITableCache {
TsTableInternalRPCUtil.deserializeTableInitializationInfo(tableInitializationBytes);
final Map<String, List<TsTable>> usingMap = tableInfo.left;
final Map<String, List<TsTable>> specialStatusMap = tableInfo.right;
+ previousTableMap.clear();
usingMap.forEach(
(key, value) ->
databaseTableMap.put(
@@ -141,6 +150,19 @@ public class DataNodeTableCache implements ITableCache {
table -> new Pair<>(table, 0L),
(v1, v2) -> v2,
ConcurrentHashMap::new))));
+ specialStatusMap.forEach(
+ (key, value) ->
+ value.stream()
+ .filter(NonCommittableTsTable.class::isInstance)
+ .forEach(
+ table ->
+ previousTableMap
+ .computeIfAbsent(
+ PathUtils.unQualifyDatabaseName(key),
+ database -> new ConcurrentHashMap<>())
+ .put(
+ table.getTableName(),
+ new
NonCommittableTsTable(table.getTableName()))));
LOGGER.info(DataNodeSchemaMessages.INIT_TABLE_CACHE_SUCCESS);
} finally {
readWriteLock.writeLock().unlock();
@@ -183,6 +205,14 @@ public class DataNodeTableCache implements ITableCache {
readWriteLock.writeLock().lock();
try {
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+ if (oldName == null && !(table instanceof PreDeleteTsTable)) {
+ final TsTable previousTable = getTableFromCache(database,
table.getTableName());
+ if (previousTable != null) {
+ previousTableMap
+ .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
+ .putIfAbsent(table.getTableName(), new TsTable(previousTable));
+ }
+ }
specialStatusMap
.computeIfAbsent(database, k -> new ConcurrentHashMap<>())
.compute(
@@ -239,13 +269,47 @@ public class DataNodeTableCache implements ITableCache {
failIfMetadataLeaseFenced(LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
// if rollback the drop table procedure, do nothing,
// wait for triggering the action of pull table from CN
- final TsTable table = getTableFromSpecialStatusMap(database, tableName);
+ final Map<String, Pair<TsTable, Long>> databaseSpecialStatusMap =
+ specialStatusMap.get(database);
+ final Pair<TsTable, Long> tableStatusPair =
+ databaseSpecialStatusMap == null ? null :
databaseSpecialStatusMap.get(tableName);
+ final TsTable table = tableStatusPair == null ? null :
tableStatusPair.getLeft();
+ final TsTable previousTable =
+ previousTableMap.containsKey(database)
+ ? previousTableMap.get(database).get(tableName)
+ : null;
if (table instanceof PreDeleteTsTable) {
+ removePreviousTable(database, tableName);
+ return;
+ }
+ // A null pending value with no saved previous schema means commit
already consumed this
+ // update. A delayed rollback must not evict the committed table.
+ if (Objects.isNull(oldName) && table == null && previousTable == null) {
return;
}
removeTableFromSpecialStatusMap(database, tableName);
+ removePreviousTable(database, tableName);
LOGGER.info(DataNodeSchemaMessages.ROLLBACK_UPDATE_TABLE_SUCCESS,
database, tableName);
+ // A table fetched while the update was pending can already be in
databaseTableMap and the
+ // special entry can consequently have a null left value. Restore the
snapshot captured at
+ // PRE_UPDATE time, or evict the entry so the next lookup must fetch the
canonical CN schema.
+ if (Objects.isNull(oldName) && tableStatusPair != null) {
+ if (previousTable != null && !(previousTable instanceof
NonCommittableTsTable)) {
+ databaseTableMap
+ .computeIfAbsent(database, k -> new ConcurrentHashMap<>())
+ .put(tableName, previousTable);
+ } else if (databaseTableMap.containsKey(database)) {
+ databaseTableMap.get(database).remove(tableName);
+ }
+ if (previousTable == null || previousTable instanceof
NonCommittableTsTable) {
+ // The previous schema is unavailable after a restart or when this
was a newly-created
+ // table. Keep a non-committable marker so getTable() fetches the
canonical CN state
+ // instead of serving a stale entry or permanently treating the
cache as already handled.
+ tableStatusPair.setLeft(new NonCommittableTsTable(tableName));
+ }
+ }
+
// If rename table
if (Objects.nonNull(oldName)) {
// Equals to commit update
@@ -313,6 +377,15 @@ public class DataNodeTableCache implements ITableCache {
});
}
+ private void removePreviousTable(final String database, final String
tableName) {
+ previousTableMap.computeIfPresent(
+ database,
+ (k, v) -> {
+ v.remove(tableName);
+ return v.isEmpty() ? null : v;
+ });
+ }
+
@Override
public void commitUpdateTable(
String database, final String tableName, final @Nullable String oldName)
{
@@ -330,6 +403,7 @@ public class DataNodeTableCache implements ITableCache {
if (Objects.nonNull(oldName)) {
removeTableFromSpecialStatusMap(database, oldName);
}
+ removePreviousTable(database, tableName);
return;
}
// Cannot be committed, consider:
@@ -359,6 +433,7 @@ public class DataNodeTableCache implements ITableCache {
LOGGER.info(DataNodeSchemaMessages.COMMIT_UPDATE_TABLE_SUCCESS,
database, tableName);
}
removeTableFromSpecialStatusMap(database, tableName);
+ removePreviousTable(database, tableName);
if (Objects.nonNull(oldName)) {
removeTableFromSpecialStatusMap(database, oldName);
LOGGER.info(DataNodeSchemaMessages.RENAME_OLD_TABLE_SUCCESS, database,
oldName);
@@ -374,6 +449,7 @@ public class DataNodeTableCache implements ITableCache {
databaseTableMap.get(database).remove(tableName);
}
removeTableFromSpecialStatusMap(database, tableName);
+ removePreviousTable(database, tableName);
LOGGER.info(DataNodeSchemaMessages.COMMIT_DELETE_TABLE_SUCCESS, database,
tableName);
}
@@ -384,6 +460,7 @@ public class DataNodeTableCache implements ITableCache {
try {
databaseTableMap.remove(database);
specialStatusMap.remove(database);
+ previousTableMap.remove(database);
instanceVersion.incrementAndGet();
} finally {
readWriteLock.writeLock().unlock();
@@ -401,6 +478,7 @@ public class DataNodeTableCache implements ITableCache {
try {
databaseTableMap.clear();
specialStatusMap.clear();
+ previousTableMap.clear();
instanceVersion.incrementAndGet();
} finally {
readWriteLock.writeLock().unlock();
@@ -708,11 +786,7 @@ public class DataNodeTableCache implements ITableCache {
instanceVersion.incrementAndGet();
}
if (targetTableIsStillDeleting) {
- throw new SemanticException(
- String.format(
- DataNodeSchemaMessages.THE_TABLE_IS_IN_PRE_DELETE_STATE,
- targetDatabase,
- targetTable));
+ throw new SemanticException(new
TableInDeletionException(targetDatabase, targetTable));
}
} finally {
readWriteLock.writeLock().unlock();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
index 959bd4b9d6a..68f2b7fc3f6 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.schema.table.TsTable;
import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
import
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern;
+import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import com.google.common.collect.ImmutableList;
@@ -65,6 +66,8 @@ public class TreeViewTest {
@Before
public void setup() {
+ MetadataLeaseManager.getInstance().updateFenceThresholdMs(Long.MAX_VALUE);
+ MetadataLeaseManager.getInstance().recoveryLeaseForTest(true);
TsTable tsTable = new TsTable(DEVICE_VIEW_TEST_TABLE);
tsTable.addProp(TsTable.TTL_PROPERTY, Long.MAX_VALUE + "");
tsTable.addProp(TreeViewSchema.TREE_PATH_PATTERN, "root.test" + ".**");
@@ -75,6 +78,8 @@ public class TreeViewTest {
@After
public void tearDown() {
DataNodeTableCache.getInstance().invalid(TREE_VIEW_DB);
+ MetadataLeaseManager.getInstance().updateFenceThresholdMs(20_000);
+ MetadataLeaseManager.getInstance().recoveryLeaseForTest(true);
}
@Test
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java
new file mode 100644
index 00000000000..354d21fa724
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableHeaderSchemaValidatorTest.java
@@ -0,0 +1,251 @@
+/*
+ * 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.relational.metadata.fetcher;
+
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.exception.table.ColumnInDeletionException;
+import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema;
+import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.TableSchema;
+import org.apache.iotdb.commons.schema.table.InsertNodeMeasurementInfo;
+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.FieldColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import
org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager;
+import
org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+import org.apache.iotdb.db.schemaengine.table.ITableCache;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.common.type.TypeFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+public class TableHeaderSchemaValidatorTest {
+ private static final String DATABASE = "pre_delete_write_test";
+ private static final String TABLE = "table1";
+ private final ITableCache cache = DataNodeTableCache.getInstance();
+ private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private final MPPQueryContext context = new MPPQueryContext(new
QueryId("pre_delete_write_test"));
+ private ClusterConfigTaskExecutor executor;
+ private TableHeaderSchemaValidator validator;
+ private boolean autoCreateSchema;
+ private boolean partialInsert;
+
+ @Before
+ public void setUp() {
+ autoCreateSchema = config.isAutoCreateSchemaEnabled();
+ partialInsert = config.isEnablePartialInsert();
+ executor = Mockito.mock(ClusterConfigTaskExecutor.class);
+ validator = new TableHeaderSchemaValidator(executor);
+ cache.invalid(DATABASE);
+ final TsTable table = new TsTable(TABLE);
+ table.addColumnSchema(
+ new FieldColumnSchema("live", TSDataType.INT32, TSEncoding.RLE,
CompressionType.LZ4));
+ table.addColumnSchema(
+ new FieldColumnSchema("field", TSDataType.INT32, TSEncoding.RLE,
CompressionType.LZ4));
+ table.addColumnSchema(new AttributeColumnSchema("attribute",
TSDataType.STRING));
+ cache.preUpdateTable(DATABASE, table, null);
+ cache.commitUpdateTable(DATABASE, TABLE, null);
+ // A DROP COLUMN invalidates the DataNode cache before removing ConfigNode
metadata.
+ cache.invalid(DATABASE, TABLE, "field");
+ cache.invalid(DATABASE, TABLE, "attribute");
+ Mockito.when(executor.getPreDeletedColumns(DATABASE, TABLE))
+ .thenReturn(new java.util.HashSet<>(Arrays.asList("field",
"attribute")));
+ }
+
+ @After
+ public void tearDown() {
+ DataNodeSchemaLockManager.getInstance().releaseReadLock(context);
+ cache.invalid(DATABASE);
+ config.setAutoCreateSchemaEnabled(autoCreateSchema);
+ config.setEnablePartialInsert(partialInsert);
+ }
+
+ @Test
+ public void testInsertReportsDeletionBeforeUnknownCategory() {
+ for (final boolean autoCreate : new boolean[] {true, false}) {
+ config.setAutoCreateSchemaEnabled(autoCreate);
+ final InsertNodeMeasurementInfo measurements = measurements("field",
null);
+ final SemanticException error =
+ assertThrows(
+ SemanticException.class,
+ () ->
+ validator.validateInsertNodeMeasurements(
+ DATABASE, measurements, context, true, null, null));
+ assertDeletion(error, "field");
+ }
+ }
+
+ @Test
+ public void testInsertRejectsPreDeletedAttribute() {
+ final SemanticException error =
+ assertThrows(
+ SemanticException.class,
+ () ->
+ validator.validateInsertNodeMeasurements(
+ DATABASE,
+ measurements("attribute", TsTableColumnCategory.ATTRIBUTE),
+ context,
+ true,
+ null,
+ null));
+ assertDeletion(error, "attribute");
+ }
+
+ @Test
+ public void testTsFileLoadRejectsPreDeletedFieldAndAttribute() {
+ for (final boolean autoCreate : new boolean[] {true, false}) {
+ config.setAutoCreateSchemaEnabled(autoCreate);
+ for (final String column : Arrays.asList("field", "attribute")) {
+ final TableSchema tableSchema =
+ new TableSchema(
+ TABLE,
+ Collections.singletonList(
+ new ColumnSchema(
+ column,
+ TypeFactory.getType(
+ column.equals("field") ? TSDataType.INT32 :
TSDataType.STRING),
+ false,
+ column.equals("field")
+ ? TsTableColumnCategory.FIELD
+ : TsTableColumnCategory.ATTRIBUTE)));
+ final SemanticException error =
+ assertThrows(
+ SemanticException.class,
+ () ->
+ validator.validateTableHeaderSchema4TsFile(
+ DATABASE, tableSchema, context, true, false, new
AtomicBoolean()));
+ assertDeletion(error, column);
+ }
+ }
+ }
+
+ @Test
+ public void testExistingColumnsDoNotFetchDeletionStatus() throws Exception {
+ validator.validateInsertNodeMeasurements(
+ DATABASE, measurements("live", TsTableColumnCategory.FIELD), context,
true, null, null);
+ validator.validateTableHeaderSchema4TsFile(
+ DATABASE,
+ new TableSchema(
+ TABLE,
+ Collections.singletonList(
+ new ColumnSchema(
+ "live",
+ TypeFactory.getType(TSDataType.INT32),
+ false,
+ TsTableColumnCategory.FIELD))),
+ context,
+ true,
+ false,
+ new AtomicBoolean());
+ Mockito.verify(executor, Mockito.never())
+ .getPreDeletedColumns(Mockito.anyString(), Mockito.anyString());
+ }
+
+ @Test
+ public void testMissingColumnStillReportsUnknownCategory() {
+ final SemanticException error =
+ assertThrows(
+ SemanticException.class,
+ () ->
+ validator.validateInsertNodeMeasurements(
+ DATABASE, measurements("missing", null), context, true,
null, null));
+ assertEquals(TSStatusCode.COLUMN_NOT_EXISTS.getStatusCode(),
error.getErrorCode());
+ }
+
+ @Test
+ public void testMissingColumnsFetchDeletionStatusOnce() {
+ config.setAutoCreateSchemaEnabled(false);
+ config.setEnablePartialInsert(true);
+ final InsertNodeMeasurementInfo measurements =
Mockito.mock(InsertNodeMeasurementInfo.class);
+ Mockito.when(measurements.getTableName()).thenReturn(TABLE);
+ Mockito.when(measurements.getMeasurementCount()).thenReturn(2);
+ Mockito.when(measurements.getColumnCategories())
+ .thenReturn(
+ new TsTableColumnCategory[] {TsTableColumnCategory.FIELD,
TsTableColumnCategory.FIELD});
+ Mockito.when(measurements.getMeasurementName(0)).thenReturn("missing1");
+ Mockito.when(measurements.getMeasurementName(1)).thenReturn("missing2");
+ validator.validateInsertNodeMeasurements(DATABASE, measurements, context,
true, null, null);
+ Mockito.verify(executor).getPreDeletedColumns(DATABASE, TABLE);
+ }
+
+ @Test
+ public void testTsFileMissingColumnsFetchDeletionStatusOnce() throws
Exception {
+ config.setAutoCreateSchemaEnabled(false);
+ config.setEnablePartialInsert(true);
+ final TableSchema tableSchema =
+ new TableSchema(
+ TABLE,
+ Arrays.asList(
+ new ColumnSchema(
+ "missing1",
+ TypeFactory.getType(TSDataType.INT32),
+ false,
+ TsTableColumnCategory.FIELD),
+ new ColumnSchema(
+ "missing2",
+ TypeFactory.getType(TSDataType.INT32),
+ false,
+ TsTableColumnCategory.FIELD)));
+
+ validator.validateTableHeaderSchema4TsFile(
+ DATABASE, tableSchema, context, true, false, new AtomicBoolean());
+
+ Mockito.verify(executor).getPreDeletedColumns(DATABASE, TABLE);
+ }
+
+ private InsertNodeMeasurementInfo measurements(
+ final String name, final TsTableColumnCategory category) {
+ final InsertNodeMeasurementInfo measurements =
Mockito.mock(InsertNodeMeasurementInfo.class);
+ Mockito.when(measurements.getTableName()).thenReturn(TABLE);
+ Mockito.when(measurements.getMeasurementCount()).thenReturn(1);
+ Mockito.when(measurements.getColumnCategories())
+ .thenReturn(new TsTableColumnCategory[] {category});
+ Mockito.when(measurements.getMeasurementName(0)).thenReturn(name);
+ Mockito.when(measurements.getType(0)).thenReturn(TSDataType.INT32);
+ return measurements;
+ }
+
+ private static void assertDeletion(final SemanticException error, final
String column) {
+ assertEquals(TSStatusCode.SEMANTIC_ERROR.getStatusCode(),
error.getErrorCode());
+ assertTrue(error.getCause() instanceof ColumnInDeletionException);
+ assertEquals(
+ new ColumnInDeletionException(DATABASE, TABLE, column).getMessage(),
+ error.getCause().getMessage());
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
index fd97da843bd..78af8f3eee0 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCacheTest.java
@@ -19,7 +19,13 @@
package org.apache.iotdb.db.schemaengine.table;
+import
org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy;
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.exception.table.TableInDeletionException;
+import org.apache.iotdb.commons.schema.table.NonCommittableTsTable;
+import org.apache.iotdb.commons.schema.table.PreDeleteTsTable;
import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil;
import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema;
import org.apache.tsfile.enums.TSDataType;
@@ -29,12 +35,18 @@ import org.junit.Assert;
import org.junit.Test;
import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.Semaphore;
public class DataNodeTableCacheTest {
private static final String DATABASE = "interrupted_fetch_database";
private static final String TABLE_CACHE_TEST_DATABASE =
"root.table_cache_test";
+ private static final String TABLE_CACHE_TEST_DATABASE_NAME =
"table_cache_test";
private static final String TABLE_NAME = "table1";
@Test
@@ -85,7 +97,99 @@ public class DataNodeTableCacheTest {
cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
- Assert.assertNull(cache.getTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME,
false));
+ Assert.assertFalse(
+ cache
+ .getTableSnapshot()
+ .getOrDefault(TABLE_CACHE_TEST_DATABASE_NAME,
Collections.emptyMap())
+ .containsKey(TABLE_NAME));
+ } finally {
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ }
+ }
+
+ @Test
+ public void rollbackAlteredTableRestoresOriginalSchema() throws Exception {
+ final ITableCache cache = DataNodeTableCache.getInstance();
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ try {
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME),
null);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+
+ final TsTable alteredTable = createTable(TABLE_NAME);
+ ((FieldColumnSchema)
alteredTable.getColumnSchema("s1")).setDataType(TSDataType.DOUBLE);
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, alteredTable, null);
+
+ // A concurrent fetch may promote the pending table into the regular
cache before rollback.
+ // Keep that path covered because rollback must still restore the
pre-update schema.
+ final Method updateUsingTable =
+ DataNodeTableCache.class.getDeclaredMethod(
+ "updateUsingTable", Map.class, Map.class,
LeaseFencedRetryPolicy.class);
+ updateUsingTable.setAccessible(true);
+ final Map<String, Map<String, TsTable>> fetchedTables = new HashMap<>();
+ fetchedTables.put(
+ TABLE_CACHE_TEST_DATABASE, Collections.singletonMap(TABLE_NAME,
alteredTable));
+ final Map<String, Map<String, Long>> previousVersions = new HashMap<>();
+ previousVersions.put(
+ TABLE_CACHE_TEST_DATABASE_NAME, Collections.singletonMap(TABLE_NAME,
1L));
+ updateUsingTable.invoke(
+ cache, fetchedTables, previousVersions,
LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+
+ cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+ Assert.assertEquals(
+ TSDataType.INT32,
+ cache
+ .getTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME)
+ .getColumnSchema("s1")
+ .getDataType());
+ } finally {
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ }
+ }
+
+ @Test
+ public void delayedRollbackDoesNotEvictCommittedAlteredSchema() {
+ final ITableCache cache = DataNodeTableCache.getInstance();
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ try {
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME),
null);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+
+ final TsTable alteredTable = createTable(TABLE_NAME);
+ ((FieldColumnSchema)
alteredTable.getColumnSchema("s1")).setDataType(TSDataType.DOUBLE);
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, alteredTable, null);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+
+ cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+ Assert.assertEquals(
+ TSDataType.DOUBLE,
+ cache
+ .getTableSnapshot()
+ .get(TABLE_CACHE_TEST_DATABASE_NAME)
+ .get(TABLE_NAME)
+ .getColumnSchema("s1")
+ .getDataType());
+ } finally {
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ }
+ }
+
+ @Test
+ public void delayedRollbackDoesNotRestoreCommittedDeletedTable() {
+ final ITableCache cache = DataNodeTableCache.getInstance();
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ try {
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, createTable(TABLE_NAME),
null);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, new
PreDeleteTsTable(TABLE_NAME), null);
+ cache.commitUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+
+ cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+ Assert.assertFalse(
+ cache
+ .getTableSnapshot()
+ .getOrDefault(TABLE_CACHE_TEST_DATABASE_NAME,
Collections.emptyMap())
+ .containsKey(TABLE_NAME));
} finally {
cache.invalid(TABLE_CACHE_TEST_DATABASE);
}
@@ -111,12 +215,97 @@ public class DataNodeTableCacheTest {
}
}
+ @Test
+ public void rollbackAfterRestartEvictsUnknownPreviousSchema() {
+ final ITableCache cache = DataNodeTableCache.getInstance();
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ try {
+ final TsTable alteredTable = createTable(TABLE_NAME);
+ ((FieldColumnSchema)
alteredTable.getColumnSchema("s1")).setDataType(TSDataType.DOUBLE);
+ final byte[] initializationBytes =
+ TsTableInternalRPCUtil.serializeTableInitializationInfo(
+ Collections.singletonMap(
+ TABLE_CACHE_TEST_DATABASE,
Collections.singletonList(alteredTable)),
+ Collections.singletonMap(
+ TABLE_CACHE_TEST_DATABASE,
+ Collections.singletonList(new
NonCommittableTsTable(TABLE_NAME))));
+ cache.init(initializationBytes);
+
+ // A restart cannot retain the in-memory pre-update snapshot. Evict the
potentially stale
+ // schema even if the recovered procedure repeats PRE_UPDATE, and fail
closed until the
+ // canonical schema can be fetched from the CN.
+ cache.preUpdateTable(TABLE_CACHE_TEST_DATABASE, alteredTable, null);
+ cache.rollbackUpdateTable(TABLE_CACHE_TEST_DATABASE, TABLE_NAME, null);
+ Assert.assertFalse(
+ cache
+ .getTableSnapshot()
+ .getOrDefault(TABLE_CACHE_TEST_DATABASE_NAME,
Collections.emptyMap())
+ .containsKey(TABLE_NAME));
+ } finally {
+ cache.invalid(TABLE_CACHE_TEST_DATABASE);
+ }
+ }
+
private Semaphore getFetchTableSemaphore(final ITableCache cache) throws
Exception {
final Field field =
DataNodeTableCache.class.getDeclaredField("fetchTableSemaphore");
field.setAccessible(true);
return (Semaphore) field.get(cache);
}
+ @Test
+ public void preDeletedTableRefreshReportsDeletionAndRecovers() throws
Exception {
+ final ITableCache cache = DataNodeTableCache.getInstance();
+ final Method updateDeleteTable =
+ DataNodeTableCache.class.getDeclaredMethod(
+ "updateDeleteTable",
+ Map.class,
+ String.class,
+ String.class,
+ LeaseFencedRetryPolicy.class);
+ updateDeleteTable.setAccessible(true);
+ final String database = "pre_delete_table_test";
+ cache.invalid(database);
+ try {
+ cache.preUpdateTable(database, new PreDeleteTsTable(TABLE_NAME), null);
+ final InvocationTargetException failure =
+ Assert.assertThrows(
+ InvocationTargetException.class,
+ () ->
+ updateDeleteTable.invoke(
+ cache,
+ Collections.singletonMap(
+ database,
+ Collections.singletonMap(TABLE_NAME, new
PreDeleteTsTable(TABLE_NAME))),
+ database,
+ TABLE_NAME,
+ LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS));
+ Assert.assertTrue(failure.getCause() instanceof SemanticException);
+ Assert.assertEquals(
+ new TableInDeletionException(database, TABLE_NAME).getMessage(),
+ failure.getCause().getCause().getMessage());
+
+ updateDeleteTable.invoke(
+ cache,
+ Collections.singletonMap(
+ database, Collections.singletonMap(TABLE_NAME,
createTable(TABLE_NAME))),
+ database,
+ TABLE_NAME,
+ LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+ Assert.assertNotNull(cache.getTableInWrite(database, TABLE_NAME));
+
+ cache.preUpdateTable(database, new PreDeleteTsTable(TABLE_NAME), null);
+ updateDeleteTable.invoke(
+ cache,
+ Collections.singletonMap(database,
Collections.singletonMap(TABLE_NAME, null)),
+ database,
+ TABLE_NAME,
+ LeaseFencedRetryPolicy.RETRY_UNTIL_SUCCESS);
+ Assert.assertNull(cache.getTableInWrite(database, TABLE_NAME));
+ } finally {
+ cache.invalid(database);
+ }
+ }
+
private TsTable createTable(final String tableName) {
final TsTable table = new TsTable(tableName);
table.addColumnSchema(
diff --git
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
index a05c8f9915a..806581d0e09 100644
---
a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -335,4 +335,10 @@ public final class CommonMessages {
public static final String
EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E =
"Snapshot buffer size must not exceed %d bytes, but was %d.";
+ public static final String
EXCEPTION_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROP_TABLE_IF_IT_IS_STUCK_7E22D78F
=
+ "Table '%s.%s' is being deleted. Please wait for deletion to finish, or
retry DROP TABLE if it is stuck.";
+ public static final String
EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE
=
+ "Column '%s' in table '%s.%s' is being deleted. Please wait for deletion
to finish, or retry dropping the column if it is stuck.";
+ public static final String
EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_ALTERED_PLEASE_WAIT_FOR_ALTERATION_TO_FINISH_OR_RETRY_ALTERING_THE_COLUMN_IF_IT_IS_STUCK_11155B55
=
+ "Column '%s' in table '%s.%s' is being altered. Please wait for
alteration to finish, or retry altering the column if it is stuck.";
}
diff --git
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
index c5c6290687e..76ce8769aa1 100644
---
a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
+++
b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java
@@ -232,4 +232,10 @@ public final class CommonMessages {
public static final String
EXCEPTION_SNAPSHOT_BUFFER_SIZE_MUST_NOT_EXCEED_ARG_BYTES_BUT_WAS_ARG_D1DA6F7E =
"快照缓冲区大小不得超过 %d 字节，但实际为 %d。";
+ public static final String
EXCEPTION_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROP_TABLE_IF_IT_IS_STUCK_7E22D78F
=
+ "表 '%s.%s' 正在删除中。请等待删除完成；如果删除一直未完成，请重试 DROP TABLE。";
+ public static final String
EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE
=
+ "列 '%s'（位于表 '%s.%s'）正在删除中。请等待删除完成；如果删除一直未完成，请重试删除该列。";
+ public static final String
EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_ALTERED_PLEASE_WAIT_FOR_ALTERATION_TO_FINISH_OR_RETRY_ALTERING_THE_COLUMN_IF_IT_IS_STUCK_11155B55
=
+ "列 '%s'（位于表 '%s.%s'）正在修改中。请等待修改完成；如果修改一直未完成，请重试修改该列。";
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java
new file mode 100644
index 00000000000..20f24abe062
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInAlterException.java
@@ -0,0 +1,40 @@
+/*
+ * 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.exception.table;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class ColumnInAlterException extends MetadataException {
+
+ public ColumnInAlterException(
+ final String database, final String tableName, final String columnName) {
+ super(
+ String.format(
+ CommonMessages
+
.EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_ALTERED_PLEASE_WAIT_FOR_ALTERATION_TO_FINISH_OR_RETRY_ALTERING_THE_COLUMN_IF_IT_IS_STUCK_11155B55,
+ columnName,
+ PathUtils.unQualifyDatabaseName(database),
+ tableName),
+ TSStatusCode.SEMANTIC_ERROR.getStatusCode());
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java
new file mode 100644
index 00000000000..b3a6e60be09
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/ColumnInDeletionException.java
@@ -0,0 +1,40 @@
+/*
+ * 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.exception.table;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class ColumnInDeletionException extends MetadataException {
+
+ public ColumnInDeletionException(
+ final String database, final String tableName, final String columnName) {
+ super(
+ String.format(
+ CommonMessages
+
.EXCEPTION_COLUMN_ARG_IN_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROPPING_THE_COLUMN_IF_IT_IS_STUCK_875DAFFE,
+ columnName,
+ PathUtils.unQualifyDatabaseName(database),
+ tableName),
+ TSStatusCode.SEMANTIC_ERROR.getStatusCode());
+ }
+}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java
new file mode 100644
index 00000000000..8ec7d18937c
--- /dev/null
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/exception/table/TableInDeletionException.java
@@ -0,0 +1,38 @@
+/*
+ * 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.exception.table;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.i18n.CommonMessages;
+import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+public class TableInDeletionException extends MetadataException {
+
+ public TableInDeletionException(final String database, final String
tableName) {
+ super(
+ String.format(
+ CommonMessages
+
.EXCEPTION_TABLE_ARG_ARG_IS_BEING_DELETED_PLEASE_WAIT_FOR_DELETION_TO_FINISH_OR_RETRY_DROP_TABLE_IF_IT_IS_STUCK_7E22D78F,
+ PathUtils.unQualifyDatabaseName(database),
+ tableName),
+ TSStatusCode.SEMANTIC_ERROR.getStatusCode());
+ }
+}
